MCP Server Development with Python SDK

MCP Server Development - with Python SDK

Introduction and Setup

This guide explains how to build MCP (Model Context Protocol) servers using the Python SDK. MCP servers are crucial components that expose Resources (data), Prompts (interfaces to models), and Tools (executable functions) to various MCP clients.

The Python SDK for MCP allows for rapid server development with a clean, decorator-based API and leverages Python’s asyncio library for efficient handling of concurrent client connections.

Prerequisites:

• Python (version 3.8 or later is recommended).
• pip (Python’s package installer).

Installation:

Install the MCP Python SDK from PyPI:

pip install modelcontext-mcp

Initializing the Server: You start by creating an instance of the MCPServer class. Then, you define handlers for your resources, prompts, and tools, typically using decorators. Finally, you start the server to listen for incoming connections.

import asyncio
from modelcontext_mcp import MCPServer, ResourceContext, PromptContext, ToolContext, MCPErrorData

# Initialize the server instance globally or within your main async function
server = MCPServer(
    server_id='my-first-python-mcp-server',
    server_description='A demonstration MCP server built with the Python SDK.',
    # Define server capabilities for clients to discover
    capabilities={
        'resources': [
            {'id': 'ping', 'description': 'A simple ping resource to check server health.'}
            # More resource capabilities can be listed here
        ],
        'prompts': [
            {'id': 'reverse_string', 'description': 'Reverses the input string.'}
            # More prompt capabilities
        ],
        'tools': [
            {'id': 'simple_math', 'description': 'Performs basic arithmetic operations.'}
            # More tool capabilities
        ]
    }
)

async def main_server_loop():
    # Handlers will be defined and registered using decorators in the next sections.
    # For now, assume they are defined elsewhere or will be added below.

    port = 8080
    host = '0.0.0.0'  # Listen on all available network interfaces

    try:
        print(f'Starting Python MCP Server on ws://{host}:{port}{server.path}...') Keating on ws://{host}:{port}{server.path}...')
        await server.start(host=host, port=port)
        print('Python MCP Server is running. Press Ctrl+C to stop.')
        print(f'Registered capabilities: {server.get_capabilities()}')
        # Keep the server running indefinitely (server.start() handles the loop)
        # If server.start() was not blocking, you would need something like:
        # while True:
        #     await asyncio.sleep(3600) # Keep alive
    except KeyboardInterrupt:
        print('\nKeyboardInterrupt received. Shutting down...')
    except Exception as e:
        print(f'Failed to start or run MCP server: {e}')
    finally:
        if server.is_running():
            print('Stopping server...') Keating server...')
            await server.stop()
            print('Server stopped.')

# To run the server (handlers will be defined using decorators below):
# if __name__ == "__main__":
#     # Define handlers here before running
#     asyncio.run(main_server_loop())

Exposing Resources

Resources provide data to clients. In the Python SDK, you define a resource handler function and decorate it with @server.resource(resource_id). The handler receives a ResourceContext object and should return a dictionary representing the payload. For streaming data, the handler can be an async generator (using async def and yield).

# Define these handlers before calling asyncio.run(main_server_loop())
# 'server' is the MCPServer instance created earlier.

# Example 1: A simple ping resource (single response)
@server.resource(resource_id='ping')
async def ping_handler(context: ResourceContext) -> dict:
    client_ip = context.client_address[0] if context.client_address else 'unknown'
    print(f"Resource 'ping' requested by client: {context.client_id} from IP: {client_ip}")
    return {'message': 'pong', 'timestamp': asyncio.get_event_loop().time()}

# Example 2: A resource that streams numbers (async generator for streaming)
@server.resource(resource_id='number_stream')
async def number_stream_handler(context: ResourceContext):
    # Params could specify start, end, step, etc.
    count = context.params.get('count', 5)
    delay = context.params.get('delay', 0.5)
    print(f"Resource 'number_stream' for {count} numbers requested by: {context.client_id}")
    for i in range(1, count + 1):
        await asyncio.sleep(delay)  # Simulate work or data fetching
        yield {'number': i, 'is_even': i % 2 == 0}
    print(f"Finished streaming for 'number_stream' for client {context.client_id}")

Exposing Prompts

Prompts are typically used for interactions that involve some form of generation or transformation, often with AI models. Define a prompt handler with @server.prompt(prompt_id). It receives PromptContext and returns a payload dictionary or yields payloads if it’s a streaming prompt.

# Define these handlers before calling asyncio.run(main_server_loop())

# Example 1: A prompt that reverses a string (single response)
@server.prompt(prompt_id='reverse_string')
async def reverse_string_handler(context: PromptContext) -> dict:
    input_text = context.params.get('text')
    print(f"Prompt 'reverse_string' by {context.client_id} with text: '{input_text}'")
    if not isinstance(input_text, str):
        raise MCPErrorData(code=400, message='"text" parameter must be a string.')
    return {'reversed_text': input_text[::-1], 'original_length': len(input_text)}

# Example 2: A mock LLM streaming prompt (async generator)
@server.prompt(prompt_id='mock_llm_stream')
async def mock_llm_stream_handler(context: PromptContext):
    prompt_text = context.params.get('prompt', 'Tell me a story.')
    words_to_stream = f"Okay, here's a story about '{prompt_text[:20]}...': Once upon a time...".split()
    print(f"Streaming prompt 'mock_llm_stream' for client {context.client_id}")
    for word in words_to_stream:
        await asyncio.sleep(0.2)  # Simulate LLM token generation delay
        yield {'token': word, 'type': 'word'}
    yield {'token': '<EOS>', 'type': 'control'}
    print(f"Finished streaming for 'mock_llm_stream' for client {context.client_id}")

Exposing Tools

Tools are server-side functions that clients can execute directly. Define a tool handler with @server.tool(tool_id). It receives ToolContext and returns a payload dictionary containing the result of the tool’s execution.

# Define these handlers before calling asyncio.run(main_server_loop())

@server.tool(tool_id='simple_math')
async def simple_math_handler(context: ToolContext) -> dict:
    operation = context.params.get('operation')
    a = context.params.get('a')
    b = context.params.get('b')
    print(f"Tool 'simple_math' ({operation}) by {context.client_id} with a={a}, b={b}")

    if not all(isinstance(x, (int, float)) for x in [a, b]):
        raise MCPErrorData(code=400, message='Parameters "a" and "b" must be numbers.')

    result: float
    if operation == 'add':
        result = a + b
    elif operation == 'subtract':
        result = a - b
    elif operation == 'multiply':
        result = a * b
    elif operation == 'divide':
        if b == 0:
            raise MCPErrorData(code=400, message='Division by zero is not allowed.')
        result = a / b
    else:
        raise MCPErrorData(code=400, message=f'Unsupported operation: {operation}. Supported: add, subtract, multiply, divide.')

    return {'result': result, 'inputs': {'a': a, 'b': b, 'operation': operation}}

Full Server Example

Here’s a complete example that includes the server initialization and all the handlers defined above. Save this as a Python file (e.g., mcp_py_server.py) and run it.

import asyncio
from modelcontext_mcp import MCPServer, ResourceContext, PromptContext, ToolContext, MCPErrorData

# 1. Initialize Server Instance
server = MCPServer(
    server_id='python-mcp-server-full',
    server_description='Full Python MCP Server Example with various capabilities.',
    capabilities={
        'resources': [
            {'id': 'ping', 'description': 'Simple health check.'},
            {'id': 'number_stream', 'description': 'Streams a sequence of numbers.'}
        ],
        'prompts': [
            {'id': 'reverse_string', 'description': 'Reverses an input string.'},
            {'id': 'mock_llm_stream', 'description': 'Simulates a streaming LLM response.'}
        ],
        'tools': [
            {'id': 'simple_math', 'description': 'Performs add, subtract, multiply, divide.'}
        ]
    }
)

# 2. Define Resource Handlers
@server.resource(resource_id='ping')
async def ping_handler(context: ResourceContext) -> dict:
    print(f"PING from {context.client_id}")
    return {'message': 'pong from Python', 'server_time': asyncio.get_event_loop().time()}

@server.resource(resource_id='number_stream')
async def number_stream_handler(context: ResourceContext):
    count = context.params.get('count', 3); delay = context.params.get('delay', 0.3)
    print(f"NUMSTREAM for {context.client_id}: count={count}")
    for i in range(1, count + 1):
        await asyncio.sleep(delay)
        yield {'current_number': i, 'is_last': i == count}

# 3. Define Prompt Handlers
@server.prompt(prompt_id='reverse_string')
async def reverse_string_handler(context: PromptContext) -> dict:
    text = context.params.get('text', '')
    print(f"REVERSE for {context.client_id}: '{text}'")
    if not isinstance(text, str): raise MCPErrorData(code=400, message='Need string for text.')
    return {'reversed': text[::-1]}

@server.prompt(prompt_id='mock_llm_stream')
async def mock_llm_stream_handler(context: PromptContext):
    text_parts = ['This ', 'is ', 'a ', 'mocked ', 'LLM ', 'stream.']
    print(f"LLMSTREAM for {context.client_id}")
    for part in text_parts:
        await asyncio.sleep(0.1)
        yield {'token': part}
    yield {'token': '<END>'}

# 4. Define Tool Handlers
@server.tool(tool_id='simple_math')
async def simple_math_handler(context: ToolContext) -> dict:
    op, a, b = context.params.get('operation'), context.params.get('a'), context.params.get('b')
    print(f"MATH for {context.client_id}: {a} {op} {b}")
    if not all(isinstance(x, (int, float)) for x in [a,b]): raise MCPErrorData(400, 'a,b num err')
    if op == 'add': result = a + b
    elif op == 'subtract': result = a - b
    elif op == 'multiply': result = a * b
    elif op == 'divide':
        if b == 0: raise MCPErrorData(400, 'div by zero')
        result = a / b
    else: raise MCPErrorData(400, f'bad op: {op}')
    return {'outcome': result}

# 5. Main Server Loop and Event Handlers (Optional)
@server.on_connect
async def handle_connect(client_id: str, client_address: tuple):
    print(f"EVENT: Client {client_id} connected from {client_address}")

@server.on_disconnect
async def handle_disconnect(client_id: str):
    print(f"EVENT: Client {client_id} disconnected")

@server.on_error
async def handle_server_error(error: Exception, client_id: str | None = None):
    print(f"EVENT: Server error occurred (Client: {client_id if client_id else 'N/A'}): {error}")

async def run_python_mcp_server():
    port = 8080
    host = '0.0.0.0'
    try:
        print(f'Starting Full Python MCP Server on ws://{host}:{port}{server.path}...')
        await server.start(host=host, port=port)
        print('Server is running. Press Ctrl+C to stop.')
        # server.start() is blocking, so this part might not be reached if it runs forever.
        # For a non-blocking start, you might await server.serve() and manage the loop.
        while True: # Keep main coroutine alive if server.start isn't fully blocking
            await asyncio.sleep(3600)
    except KeyboardInterrupt:
        print('\nCtrl+C pressed. Shutting down the server gracefully...') except Exception as e:
        print(f'An unexpected error occurred while running the server: {e}')
    finally:
        if server.is_running():
            await server.stop()
        print('Python MCP Server has been shut down.')

if __name__ == "__main__":
    try:
        asyncio.run(run_python_mcp_server())
    except KeyboardInterrupt:
        print("Application terminated by user.")

Best Practices for Python MCP Servers

• Decorator-Based Handlers: Utilize the @server.resource, @server.prompt, and @server.tool decorators for a clean and declarative way to define your server’s capabilities.
• Asynchronous Operations: Ensure all handler functions that perform I/O (file access, network calls, database queries, LLM interactions) are defined as async def and use await for non-blocking operations.
• Type Hinting: Use Python type hints in your handler signatures (e.g., context: ResourceContext, -> dict) for better code readability and to aid static analysis tools, even though they are not strictly enforced at runtime by default.
• Input Validation: Rigorously validate parameters from context.params in each handler. Raise MCPErrorData(code=..., message=...) for client errors (e.g., missing parameters, invalid types).
• Streaming with Async Generators: For resources or prompts that need to stream multiple payloads, implement the handler as an async def function that yields each payload dictionary.
• Contextual Information: Leverage the context object (e.g., context.client_id, context.params, context.client_address) for logging, personalized responses, or authorization logic.
• Clear Capabilities: Define comprehensive server capabilities during MCPServer instantiation so clients can accurately discover what your server offers.
• Graceful Shutdown: Handle KeyboardInterrupt and other signals to call await server.stop() for a clean shutdown, allowing connections to terminate properly.
• Event Handlers: Use @server.on_connect, @server.on_disconnect, and @server.on_error decorators to implement custom logic for these server lifecycle events, such as logging or resource cleanup.