Agentics with the Model Context Protocol and Pure Storage

This is part 2 of my blog series on integrating Pure Storage on-premises full flash arrays with the Claude.ai Model Context Protocol. In part 1 — see here- I covered the code required to create the basic integration allowing for hardcoded calls from the MCP tools to a given array, but I want to be able to to take actions as well as retrieve information and not end up with one MCP tool per request.

To accomplish the above I need to change from a hardcoded restAPI endpoint call on the storage array within the MCP server tool, like get_arrays used in part 1 blog’ pure-info implementation, to a modular approach where it accepts any endpoint call as a parameter. The following code implements this and when setup it provides a single ‘pure-fb’ agent:

Code

Here is the modified code that allows us to pass queries with parameters against a given array identified by host and api-token. The new MCP server tool is named pure-fb and currently leverages token authorization. The full code can be found in my github repository:

import asyncio

from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
from pydantic import AnyUrl
import mcp.server.stdio

import json
import pypureclient
from pypureclient import flashblade

from datetime import datetime, timedelta

server = Server("pure-mcp-server")

class FlashbladeClient:
    """
    A client for interacting with Pure Storage Flashblade REST API 2.0
    """
    
    def __init__(self, fb_host, fb_api_token):
        """
        Initialize the Flashblade client
        
        Args:
            fb_host (str): The hostname or IP address of the Flashblade
            fb_api_token (str): The API token for authentication
        """
        try:
            self.client = flashblade.Client(target=fb_host, api_token=fb_api_token, verify_ssl=False, user_agent='MCP/0.1.0')
        except pypureclient.exceptions.PureError as e:
            print(f"Exception when logging in to the array: {e}")

    def call_endpoint(self, method_name, **kwargs):
        """
        Calls a method on the underlying flashblade.Client object by name.

        Args:
            method_name (str): The name of the method to call on the client.
            **kwargs: Any keyword arguments to pass to that method.

        Returns:
            The result of the underlying client method call, or None on error.
        """
        try:
            method = getattr(self.client, method_name)
            return method(**kwargs)
        except pypureclient.exceptions.PureError as e:
            print(f"Exception with {method_name}: {e}")
            return None
        except AttributeError:
            print(f"Method '{method_name}' not found on flashblade.Client")
            return None


def json_log(response, endpoint_name):
    """
    Serializes and logs the response from a Flashblade API endpoint.

    Args:
        response: The API response to be logged and serialized.
        endpoint_name (str): The name of the API endpoint.

    Returns:
        str: JSON-formatted string of the response data.
    """
    if isinstance(response, pypureclient.responses.ValidResponse):
        if response and hasattr(response, "items"):
            json_data = json.dumps([item.to_dict() for item in response.items])
        else:
            json_data = json.dumps({"error": f"No data available for {endpoint_name}"})
    else:
        json_data = json.dumps({"error": f"Invalid response from {endpoint_name}"})

    server.request_context.session.send_log_message(
        level="info",
        data=f"{endpoint_name}: {json_data}"
    )
    return json_data

### MCP tools ###
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """
    List available tools.
    Each tool specifies its arguments using JSON schema validation.
    """
    return [
        types.Tool(
            name="pure-fb",
            description="Run a command against a given FlashBlade",
            inputSchema={
                "type": "object",
                "properties": {
                    "host": {"type": "string", "description": "IP address of array management endpoint"},
                    "api_token": {"type": "string", "description": "API token for array management user"},
                    "command": {"type": "string", "description": "SDK call to run against the array"},
                    "parameters": {
                        "type": "object",
                        "description": "Optional parameters to pass to the SDK call",
                        "additionalProperties": True
                    }
                },
                "required": ["host", "api_token", "command"],
            },
        ),
    ]


@server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    Handle tool execution requests.
    """
    if name not in ["pure-fb"]:
        raise ValueError(f"Unknown tool: {name}")

    if not arguments:
        raise ValueError("Missing arguments")

    query_host = arguments.get("host")
    query_api_token = arguments.get("api_token")

    server.request_context.session.send_log_message(
        level="info",
        data=f"Initializing query with '{query_host}' and '{query_api_token}'",
    )

    if not query_host or not query_api_token:
        raise ValueError("Missing host or api_token")

    # Init array client
    fb_client = FlashbladeClient(fb_host=query_host, fb_api_token=query_api_token)

    if name == "pure-fb":
        query_command = arguments.get("command")
        query_parameters = arguments.get("parameters", {})

        if not query_command:
            raise ValueError("Missing command")

        try:
            response = fb_client.call_endpoint(query_command, **query_parameters)
            response_info = json_log(response, query_command)
            return [types.TextContent(type="text", text=f"{response_info}")]
        except Exception as error:
            return types.CallToolResult(
                isError=True,
                content=[types.TextContent(type="text", text=f"Error: {str(error)}")]
            )

async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="pure-mcp-server",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

Usage

With our new MCP server in place I can run informational queries such as :

“use pure-fb on host: 1.2.3.4 with api-token: T-7c7a*****f23e to run the command: get_file_systems and then the command: get_buckets. Sort the items per used size and indicate for each if it is a bucket by tagging the item as ‘s3’ or a file system by tagging the item as ‘fs’. Make sure to correctly sort by descending order based on used size and list the top 5 by used space.”

This results in the following output:

I'll help you retrieve and analyze the file systems and buckets information 
from your FlashBlade.
View result from pure-fb from pureflashblade (local){}
View result from pure-fb from pureflashblade (local){}

Here are the top 5 storage items (combining both file systems and buckets) 
sorted by used space in descending order:
lbo-outscale-perfs [fs]: 7,240,575,042,227 bytes (7.24 TB)
lbo-vertica [fs]: 3,311,892,804,361 bytes (3.31 TB)
dm [fs]: 2,213,617,471,583 bytes (2.21 TB)
lbo-hdf5 [fs]: 1,643,315,388,928 bytes (1.64 TB)
eg-px-share01 [fs]: 869,598,847,641 bytes (869.6 GB)
As we can see, the top 5 spots are all occupied by file systems, with the 
largest being lbo-outscale-perfs at 7.24 TB. For reference, the largest 
bucket (cvlt-uklab-object-lock at 265.1 GB) would rank much lower in this 
combined list.

This is similar to what I showed in part 1 but the modular approach now allows me to query any endpoint (get arrays, buckets, file systems,…). This modular approach also enables calling endpoints to take actions on the array, for this I will run a simple create bucket command using the following query:

“use pure-fb on host: 1.2.3.4 with api-token: T-7c7a*****f23e and the command post_buckets with the parameters: names=[“jbtmcptest”], bucket=BucketPost(account=Reference(name=’jbt’)) ”

On the Pure Storage FlashBlade array the relevant bucket is now present:

and an example query to create a file system, note that in this example I let the LLM figure out the actual command:

“create a file system using the following parameters: names=[“jbt-myfs”], file_system=FileSystemPost(provisioned=5000, hard_limit_enabled=True, nfs=Nfs(v4_1_enabled=True), smb=Smb(enabled=True), multi_protocol=MultiProtocolPost(access_control_style=”nfs”))”

We can see that ClaudeAI was able to provide the correct command ‘post_file_system’ to run this request !

Agent

I will now add some context to my chat and define the behaviour and capabilities of a storage agent that has two arrays to control:

“You are a smart storage management agent that via the pure-fb tool can manage the following two arrays:

array01 at host: 1.2.3.4 with api-token: T-7c7a*****f23e

array02 at host: 1.2.3.5 with api-token: T-f2e0*****3b8d”

With this storage agent defined I can run queries against both arrays such as:

“Get all buckets from both arrays, extract name, raw total_used space, and source array into a single list. Sort strictly based on the raw total_used values without any conversions. Provide the top 15 from the list and convert used space to appropriate units for display.”

Our agent starts by gathering the relevant data using get_buckets and get_file_systems against the two arrays:

Then the agent proceeds to sort the buckets by used space as requested. Note that it took several attempts to get the prompt for the sorting to be correct.

Conclusion

With our Model Context Protocol server we enabled natural language integration between an on-premises Pure Storage array and our LLM, to run queries and take actions.

Agentics is enabling the “new possible” and will drive an interesting next couple of years !