Unlocking On-Premises Storage Potential with the Model Context Protocol (MCP)

In today’s data-driven landscape, efficient and secure storage management is critical for enterprises handling vast amounts of information. Reporting and collecting information to understand our data footprint has long been a tedious task, but what if we could leverage a natural language interface to query information from our hardware to facilitate insights, and if wanted expand this capability into taking actions.

MCP is an open protocol that standardises how applications provide context to LLMs. This blog will cover the steps to build an MCP server to query Pure Storage arrays (FlashBlade specifically), from within Claude desktop and get a natural language response as per this screenshot:

Code

First step is to install the uv python package manager, the official guide is here: https://docs.astral.sh/uv/getting-started/installation/#installation-methods on my macbook I used brew:

% brew install uv

Next we need to create the MCP server structure, for this we will make use of the provided tools: uvx create-mcp-server.

% uvx create-mcp-server     
Creating a new MCP server project using uv.
This will set up a Python project with MCP dependency.

Let's begin!

Project name (required): pure-mcp-server
Project description [A MCP server project]: A MCP server to retrieve realtime information from a Pure Storage FlashBlade
Project version [0.1.0]: 
Project will be created at: /Users/jthomas/mcp/pure-mcp-server
Is this correct? [Y/n]: Y
....
Claude.app detected. Would you like to install the server into Claude.app now? [Y/n]: Y

✅ Added pure-mcp-server to Claude.app configuration
Settings file location: /Users/jthomas/Library/Application Support/Claude/claude_desktop_config.json
✅ Created project pure-mcp-server in pure-mcp-server
ℹ️ To install dependencies run:
   cd pure-mcp-server
   uv sync --dev --all-extras

Proceed with the final steps to install dependencies and our skeleton MCP server is ready. The source code that we need to modify is located under pure-mcp-server/src/pure-mcp-server at this point the code is for the template add-notes MCP server, we will now modify that so as to integrate with an onpremise Pure Storage FlashBlade array.

From the project folder I enter into the python virtual environment that was automatically created by uv.

% source .venv/bin/activate
(pure-mcp-server) %

Then, install the pure storage python SDK, as I will leverage this to interact with the FlashBlade restAPI.

% uv pip install py-pure-client

Within the src/pure-mcp-server/server.py file I add this library to our import section :

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")

Then I create a FlashBlade class to hold all the SDK calls that we will call to from the MCP tool definitions. To simplify the code, I am using a call_endpoint definition to pass the relevant restAPI endpoint as required.

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:
            host (str): The hostname or IP address of the Flashblade
            api_token (str): The API token for authentication
            verify_ssl (bool): Whether to verify SSL certificates
        """
        try:
            # login to the array with your API_TOKEN
            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("Exception when logging in to the array: %s\n" % e)

    def call_endpoint(self, method_name, **kwargs):
        """
        Calls a method on the underlying flashblade.Client object by name,
        catching and logging any PureError exceptions.

        Args:
            method_name (str): The name of the method to call on the client.
                               (e.g., 'get_admins', 'get_alert_watchers', etc.)
            **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

    # Example usage:
    # fb_client = FlashbladeClient("1.2.3.4", "your_api_token")
    # response = fb_client.call_endpoint("get_admins", username="pureuser", expose_api_token=True)

I added a helper function to streamline the api response to json output:

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.
    """
    # Validate response type
    if isinstance(response, pypureclient.responses.ValidResponse):
        # Check if the response has items and serialize them
        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}"})

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

    return json_data

Within the tools section I define the MCP “tools” that we will be able to run later from within Claude desktop app to the MCP server:

@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="get-array",
            description="Get array information",
            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",
                    },
                },
                "required": ["host","api_token"],
            },
        ),
        types.Tool(
            name="get-array-full",
            description="Get array information, space and 7 days performance",
            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",
                    },
                },
                "required": ["host","api_token"],
            },
        ),
    ]

I then handle the call to each tool definition, I have only added two simple calls but the example framework provides an easy method to add whatever calls/tools are needed:

@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    """
    Handle tool execution requests.
    Tools can modify server state and notify clients of changes.
    """
    if name not in ["get-array", "get-array-full"]:
        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"initialize 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 == "get-array":
        try:
            arrays_info_response = fb_client.call_endpoint("get_arrays")
            arrays_info = json_log(arrays_info_response, "get_arrays")


            return [
                types.TextContent(
                    type="text",
                    text=f"Arrays information: {arrays_info}"
                )
            ]
        except Exception as error:
            return types.CallToolResult(
                isError=True,
                content=[
                    types.TextContent(
                        type="text",
                        text=f"Error: {str(error)}"
                    )
                ]
            )
    
    elif name == "get-array-full":
        try:
            arrays_info_response = fb_client.call_endpoint("get_arrays")
            arrays_info = json_log(arrays_info_response, "get_arrays")

            arrays_space_response = fb_client.call_endpoint("get_arrays_space")
            arrays_space = json_log(arrays_space_response, "get_arrays_space")
            
            # Set the current time in epoch format (milliseconds)
            e_end_time = int(datetime.now().timestamp() * 1000)

            # Set the start time to 7 days before the current time in epoch format (milliseconds)
            e_start_time = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)

            # Define the resolution in milliseconds
            #e_resolution = 30000 * 1000  # converting 30,000 seconds to milliseconds
            
            # Corrected call with arguments passed explicitly
            arrays_performance_response = fb_client.call_endpoint(
                "get_arrays_performance", 
                start_time=e_start_time, 
                end_time=e_end_time
            )
            arrays_performance = json_log(arrays_performance_response, "get_arrays_performance")

            # Return results as MCP JSON content
            return [
                types.TextContent(
                    type="text",
                    text=f"Arrays information: {arrays_info}, space: {arrays_space}, and performance: {arrays_performance}"
                )
            ]
        except Exception as error:
            return types.CallToolResult(
                isError=True,
                content=[
                    types.TextContent(
                        type="text",
                        text=f"Error: {str(error)}"
                    )
                ]
            )

Then at last the main definition:

async def main():
    # Run the server using stdin/stdout streams
    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={},
                ),
            ),
        )

The __init__.py file contents is:

from . import server
import asyncio
import mcp.server.stdio

def main():
    """Main entry point for the package."""
    # Run both WebSocket and STDIO servers concurrently
    asyncio.run(server.main())

# Optionally expose other important items at package level
__all__ = ['main', 'server']

The MCP server was added automatically to Claude desktop during the initialisation phase, to verify open the following file and check the entry is correct:

cat /Users/jthomas/Library/Application\ Support/Claude/claude_desktop_config.json
{
  "mcpServers": {
     "pure-mcp-server": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/folder/pure-mcp-server",
        "run",
        "pure-mcp-server"
      ]
    }
  }
}

In addition to the MCP inspector I used the following script to validate:

import asyncio
from mcp.client.session import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client


async def main():
    async with stdio_client(
        StdioServerParameters(command="uv", args=["--directory", "/Users/jthomas/mcp/pure-mcp-server", "run", "pure-mcp-server"])
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print(tools)

            # Call the fetch tool
            result = await session.call_tool("get-array", {"host": "1.2.3.4", "api_token": "T-da03****1a98"})
            print(result)


asyncio.run(main())

This code correctly lists my MCP server tools, and also connects to the remote array, performs the necessary api call and returns the relevant json string.

Now I can via the Claude desktop app run queries using the MCP server tools against a given Pure Storage FlashBlade:

and using the get-array-full tool to ask about storage space specifics:

follow up questions can be asked such as : “create a graph to show the last 7 days performance based on the per metric epoch time value. Convert epoch and group performance results per day”. Note this is a pretty idle lab array:

Conclusion

The Claude.ai model context protocol provides an interesting approach to integrate systems into LLMs. The above example showed how simple it was to create an MCP server that pulls information from a Pure Storage Flashblade and uses the LLM to process the returned data. The MCP server could be expanded upon to include tools that take actions, opening the door to LLM based array management via a conversational interface.

Stay tuned for a future blog where I will cover that angle!