Model Context Protocol Python SDK (modelcontextprotocol/python-sdk)

Model Context Protocol Python SDK

https://github.com/modelcontextprotocol/python-sdk
Admin
The MCP Python SDK implements the Model Context Protocol, enabling applications to provide...

Tokens:51,884
Snippets:217
Trust Score:7.8
Update:6 hours ago
Show doc for...
Context Summary (auto-generated)
Raw
# MCP Python SDK

The Model Context Protocol (MCP) Python SDK provides a standardized framework for building servers that expose data and functionality to Large Language Model (LLM) applications. It enables seamless communication between AI applications and external data sources through a well-defined protocol supporting tools (executable functions), resources (contextual data), and prompts (reusable templates). The SDK implements the full MCP specification with support for multiple transport mechanisms including stdio, Server-Sent Events (SSE), and Streamable HTTP.

The SDK offers two development approaches: a high-level `MCPServer` class with decorator-based registration for rapid development, and a low-level `Server` class for advanced use cases requiring fine-grained control. Both client and server implementations are fully async-compatible, built on anyio for cross-platform async support, and integrate with modern Python web frameworks like Starlette for HTTP-based transports.

## Creating an MCP Server with MCPServer

The `MCPServer` class provides a high-level, decorator-based interface for building MCP servers. It handles connection management, protocol compliance, and message routing automatically.

```python
from mcp.server.mcpserver import MCPServer

# Create an MCP server with configuration
mcp = MCPServer(
    name="Demo",
    title="Demo Server",
    description="A demonstration MCP server",
    instructions="Use this server for demo purposes",
    debug=True,
    log_level="INFO"
)

# Add an addition tool using decorator
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

# Add a prompt template
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    """Generate a greeting prompt"""
    styles = {
        "friendly": "Please write a warm, friendly greeting",
        "formal": "Please write a formal, professional greeting",
        "casual": "Please write a casual, relaxed greeting",
    }
    return f"{styles.get(style, styles['friendly'])} for someone named {name}."

# Run with streamable HTTP transport (recommended for production)
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="127.0.0.1", port=8000)
```

## Registering Tools with @mcp.tool()

Tools are functions exposed to LLMs that can perform computations and have side effects. They support automatic parameter extraction, type validation, and structured output.

```python
from mcp.server.mcpserver import MCPServer, Context
from pydantic import BaseModel, Field

mcp = MCPServer(name="Tool Example")

# Basic tool with simple return type
@mcp.tool()
def sum(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

# Tool with default parameters
@mcp.tool()
def get_weather(city: str, unit: str = "celsius") -> str:
    """Get weather for a city."""
    return f"Weather in {city}: 22 degrees {unit[0].upper()}"

# Tool with Context for logging and progress reporting
@mcp.tool()
async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:
    """Execute a task with progress updates."""
    await ctx.info(f"Starting: {task_name}")

    for i in range(steps):
        progress = (i + 1) / steps
        await ctx.report_progress(
            progress=progress,
            total=1.0,
            message=f"Step {i + 1}/{steps}",
        )
        await ctx.debug(f"Completed step {i + 1}")

    return f"Task '{task_name}' completed"

# Structured output with Pydantic models
class WeatherData(BaseModel):
    """Weather information structure."""
    temperature: float = Field(description="Temperature in Celsius")
    humidity: float = Field(description="Humidity percentage")
    condition: str
    wind_speed: float

@mcp.tool()
def get_weather_data(city: str) -> WeatherData:
    """Get weather for a city - returns structured data."""
    return WeatherData(
        temperature=22.5,
        humidity=45.0,
        condition="sunny",
        wind_speed=5.2,
    )
```

## Registering Resources with @mcp.resource()

Resources expose data to LLMs similar to GET endpoints in a REST API. They provide contextual information without side effects.

```python
from mcp.server.mcpserver import MCPServer

mcp = MCPServer(name="Resource Example")

# Static resource with fixed URI
@mcp.resource("config://settings")
def get_settings() -> str:
    """Get application settings."""
    return """{
  "theme": "dark",
  "language": "en",
  "debug": false
}"""

# Dynamic resource template with URI parameters
@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
    """Read a document by name."""
    # This would normally read from disk
    return f"Content of {name}"

# Resource returning binary data
@mcp.resource("image://logo")
def get_logo() -> bytes:
    """Get the application logo."""
    with open("logo.png", "rb") as f:
        return f.read()

# Async resource for I/O operations
@mcp.resource("api://users/{user_id}")
async def get_user_data(user_id: str) -> str:
    """Get user data from external API."""
    # Simulated async data fetch
    return f'{{"id": "{user_id}", "name": "John Doe", "email": "john@example.com"}}'
```

## Registering Prompts with @mcp.prompt()

Prompts are reusable templates that help LLMs interact with the server effectively. They can return simple strings or structured message lists.

```python
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.prompts import base

mcp = MCPServer(name="Prompt Example")

# Simple prompt returning a string
@mcp.prompt(title="Code Review")
def review_code(code: str) -> str:
    return f"Please review this code:\n\n{code}"

# Prompt returning structured messages
@mcp.prompt(title="Debug Assistant")
def debug_error(error: str) -> list[base.Message]:
    return [
        base.UserMessage("I'm seeing this error:"),
        base.UserMessage(error),
        base.AssistantMessage("I'll help debug that. What have you tried so far?"),
    ]

# Prompt with multiple optional parameters
@mcp.prompt(title="SQL Query Generator")
def generate_sql(table_name: str, operation: str = "SELECT", conditions: str = "") -> str:
    """Generate SQL query based on parameters"""
    base_query = f"{operation} * FROM {table_name}"
    if conditions:
        base_query += f" WHERE {conditions}"
    return f"Generate a production-ready SQL query based on: {base_query}"
```

## Server Lifespan Management

The lifespan API allows initialization of resources at server startup and cleanup at shutdown, with type-safe context passing to handlers.

```python
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server.mcpserver import Context, MCPServer

# Mock database class
class Database:
    @classmethod
    async def connect(cls) -> "Database":
        print("Database connected")
        return cls()

    async def disconnect(self) -> None:
        print("Database disconnected")

    def query(self, sql: str) -> str:
        return f"Result for: {sql}"

@dataclass
class AppContext:
    """Application context with typed dependencies."""
    db: Database

@asynccontextmanager
async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
    """Manage application lifecycle with type-safe context."""
    # Initialize on startup
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        # Cleanup on shutdown
        await db.disconnect()

# Pass lifespan to server
mcp = MCPServer("My App", lifespan=app_lifespan)

# Access type-safe lifespan context in tools
@mcp.tool()
def query_db(sql: str, ctx: Context[AppContext]) -> str:
    """Tool that uses initialized database resources."""
    db = ctx.request_context.lifespan_context.db
    return db.query(sql)
```

## Creating MCP Clients with ClientSession

The `ClientSession` class provides a high-level interface for connecting to MCP servers using various transports.

```python
import asyncio
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

# Configure server parameters for stdio connection
server_params = StdioServerParameters(
    command="python",
    args=["my_server.py"],
    env={"DEBUG": "true"},
)

async def run_client():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print(f"Available tools: {[t.name for t in tools.tools]}")

            # Call a tool
            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
            if result.content and isinstance(result.content[0], types.TextContent):
                print(f"Tool result: {result.content[0].text}")

            # Access structured content if available
            if result.structured_content:
                print(f"Structured result: {result.structured_content}")

            # List and get prompts
            prompts = await session.list_prompts()
            print(f"Available prompts: {[p.name for p in prompts.prompts]}")

            if prompts.prompts:
                prompt = await session.get_prompt(
                    "greet_user",
                    arguments={"name": "Alice", "style": "friendly"}
                )
                print(f"Prompt result: {prompt.messages[0].content}")

            # List and read resources
            resources = await session.list_resources()
            print(f"Available resources: {[r.uri for r in resources.resources]}")

            resource_content = await session.read_resource("greeting://World")
            if resource_content.contents:
                content = resource_content.contents[0]
                if hasattr(content, 'text'):
                    print(f"Resource content: {content.text}")

if __name__ == "__main__":
    asyncio.run(run_client())
```

## Streamable HTTP Client Connection

The Streamable HTTP transport is the recommended approach for production deployments, supporting stateless operation, resumability, and better scalability.

```python
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

async def main():
    # Connect to a streamable HTTP server
    async with streamable_http_client("http://localhost:8000/mcp") as (read_stream, write_stream):
        # Create a session using the client streams
        async with ClientSession(read_stream, write_stream) as session:
            # Initialize the connection
            await session.initialize()

            # List available tools
            tools = await session.list_tools()
            print(f"Available tools: {[tool.name for tool in tools.tools]}")

            # Call a tool with progress callback
            async def progress_handler(progress: float, total: float | None, message: str | None):
                print(f"Progress: {progress}/{total} - {message}")

            result = await session.call_tool(
                "long_running_task",
                arguments={"task_name": "Analysis", "steps": 10},
                progress_callback=progress_handler,
                read_timeout_seconds=60.0
            )
            print(f"Result: {result.content}")

if __name__ == "__main__":
    asyncio.run(main())
```

## Running Server with Different Transports

The MCP server supports multiple transport protocols: stdio for subprocess communication, SSE for server-sent events, and Streamable HTTP for production deployments.

```python
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Multi-Transport Server")

@mcp.tool()
def echo(message: str) -> str:
    """Echo a message back."""
    return f"Echo: {message}"

# Option 1: Run with stdio transport (for subprocess communication)
# mcp.run(transport="stdio")

# Option 2: Run with SSE transport
# mcp.run(
#     transport="sse",
#     host="127.0.0.1",
#     port=8000,
#     sse_path="/sse",
#     message_path="/messages/"
# )

# Option 3: Run with Streamable HTTP transport (recommended for production)
if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=8000,
        streamable_http_path="/mcp",
        json_response=True,      # Use JSON responses instead of SSE
        stateless_http=True,     # Enable stateless mode for scalability
    )
```

## Mounting MCP Server in Starlette Application

For integration with existing web applications, MCP servers can be mounted into Starlette or other ASGI-compatible frameworks.

```python
import contextlib
from starlette.applications import Starlette
from starlette.routing import Mount
from mcp.server.mcpserver import MCPServer

# Create multiple MCP servers
echo_mcp = MCPServer(name="EchoServer")
math_mcp = MCPServer(name="MathServer")

@echo_mcp.tool()
def echo(message: str) -> str:
    """A simple echo tool"""
    return f"Echo: {message}"

@math_mcp.tool()
def add_two(n: int) -> int:
    """Tool to add two to the input"""
    return n + 2

# Create a combined lifespan to manage both session managers
@contextlib.asynccontextmanager
async def lifespan(app: Starlette):
    async with contextlib.AsyncExitStack() as stack:
        await stack.enter_async_context(echo_mcp.session_manager.run())
        await stack.enter_async_context(math_mcp.session_manager.run())
        yield

# Create the Starlette app and mount the MCP servers
app = Starlette(
    routes=[
        Mount("/echo", echo_mcp.streamable_http_app()),
        Mount("/math", math_mcp.streamable_http_app()),
    ],
    lifespan=lifespan,
)

# Clients connect to http://localhost:8000/echo/mcp and http://localhost:8000/math/mcp
```

## Low-Level Server API

For advanced use cases requiring fine-grained control, the low-level `Server` class provides direct access to the protocol with handler-based registration.

```python
import asyncio
from typing import Any

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

# Create a low-level server instance
server = Server("example-server")

# Register handlers using decorators
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """List available tools."""
    return [
        types.Tool(
            name="calculate",
            description="Perform a calculation",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Math expression"}
                },
                "required": ["expression"],
            },
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict[str, Any]) -> list[types.TextContent]:
    """Handle tool calls."""
    if name == "calculate":
        expression = arguments.get("expression", "")
        try:
            result = eval(expression)  # In production, use a safe evaluator
            return [types.TextContent(type="text", text=str(result))]
        except Exception as e:
            return [types.TextContent(type="text", text=f"Error: {e}")]
    raise ValueError(f"Unknown tool: {name}")

@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
    """List available prompts."""
    return [
        types.Prompt(
            name="example-prompt",
            description="An example prompt template",
            arguments=[
                types.PromptArgument(name="arg1", description="Example argument", required=True)
            ],
        )
    ]

@server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict[str, str] | None) -> types.GetPromptResult:
    """Get a specific prompt by name."""
    if name != "example-prompt":
        raise ValueError(f"Unknown prompt: {name}")

    arg1_value = (arguments or {}).get("arg1", "default")
    return types.GetPromptResult(
        description="Example prompt",
        messages=[
            types.PromptMessage(
                role="user",
                content=types.TextContent(type="text", text=f"Argument value: {arg1_value}"),
            )
        ],
    )

async def run():
    """Run the low-level server with stdio transport."""
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="example",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

if __name__ == "__main__":
    asyncio.run(run())
```

## Context Object for Tool Functions

The `Context` object provides access to MCP capabilities within tool and resource functions, including logging, progress reporting, resource access, and elicitation.

```python
from mcp.server.mcpserver import Context, MCPServer
from pydantic import BaseModel

mcp = MCPServer(name="Context Example")

# Schema for user input elicitation
class BookingPreferences(BaseModel):
    check_alternative: bool
    alternative_date: str = "2024-12-26"

@mcp.tool()
async def process_with_context(data: str, ctx: Context) -> str:
    """Demonstrate all Context capabilities."""

    # Logging at different levels
    await ctx.debug(f"Debug: Processing '{data}'")
    await ctx.info("Info: Starting processing")
    await ctx.warning("Warning: This is experimental")
    await ctx.error("Error: (This is just a demo)")

    # Log with extra structured data
    await ctx.info("Processing data", extra={"data_length": len(data), "timestamp": "2024-01-01"})

    # Report progress for long operations
    for i in range(5):
        await ctx.report_progress(
            progress=i + 1,
            total=5,
            message=f"Step {i + 1} of 5"
        )

    # Read another resource
    resource_data = await ctx.read_resource("config://settings")

    # Access request metadata
    request_id = ctx.request_id
    client_id = ctx.client_id  # May be None

    # Elicit additional information from user (form mode)
    result = await ctx.elicit(
        message="Would you like to check another date?",
        schema=BookingPreferences,
    )

    if result.action == "accept" and result.data:
        return f"User chose: {result.data.alternative_date}"

    return f"Processed: {data}"
```

## Server Authentication with OAuth

MCP servers can implement OAuth 2.1 authentication to protect resources. The SDK provides built-in support for token verification.

```python
from pydantic import AnyHttpUrl
from mcp.server.auth.provider import AccessToken, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.mcpserver import MCPServer

class SimpleTokenVerifier(TokenVerifier):
    """Simple token verifier for demonstration."""

    async def verify_token(self, token: str) -> AccessToken | None:
        # Implement actual token validation logic here
        # Return AccessToken if valid, None if invalid
        if token == "valid-token":
            return AccessToken(
                token=token,
                client_id="client-123",
                scopes=["user", "read"],
                expires_at=None
            )
        return None

# Create MCPServer as a Resource Server with authentication
mcp = MCPServer(
    "Weather Service",
    # Token verifier for authentication
    token_verifier=SimpleTokenVerifier(),
    # Auth settings for RFC 9728 Protected Resource Metadata
    auth=AuthSettings(
        issuer_url=AnyHttpUrl("https://auth.example.com"),  # Authorization Server URL
        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # This server's URL
        required_scopes=["user"],
    ),
)

@mcp.tool()
async def get_weather(city: str = "London") -> dict[str, str]:
    """Get weather data for a city (requires authentication)"""
    return {
        "city": city,
        "temperature": "22",
        "condition": "Partly cloudy",
        "humidity": "65%",
    }

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

## Custom HTTP Routes

MCP servers can expose additional HTTP endpoints alongside the MCP protocol for health checks, OAuth callbacks, or admin APIs.

```python
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Server with Custom Routes")

@mcp.tool()
def greet(name: str) -> str:
    """Greet a user."""
    return f"Hello, {name}!"

# Custom route for health checks (does not require auth)
@mcp.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> Response:
    return JSONResponse({"status": "ok", "server": mcp.name})

# Custom route for metrics
@mcp.custom_route("/metrics", methods=["GET"])
async def metrics(request: Request) -> Response:
    return JSONResponse({
        "tools_count": len(await mcp.list_tools()),
        "resources_count": len(await mcp.list_resources()),
    })

# Custom OAuth callback route
@mcp.custom_route("/oauth/callback", methods=["GET", "POST"])
async def oauth_callback(request: Request) -> Response:
    code = request.query_params.get("code")
    return JSONResponse({"status": "callback received", "code": code})
```

## ClientSession Methods Reference

The `ClientSession` class provides comprehensive methods for interacting with MCP servers, including tool calls, resource operations, and prompt management.

```python
import asyncio
from mcp import ClientSession, types
from mcp.client.streamable_http import streamable_http_client

async def demonstrate_session_methods():
    async with streamable_http_client("http://localhost:8000/mcp") as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize connection (required first)
            init_result = await session.initialize()
            print(f"Server: {init_result.server_info.name}")
            print(f"Protocol: {init_result.protocol_version}")

            # Ping the server
            await session.send_ping()

            # Tool operations
            tools = await session.list_tools()
            result = await session.call_tool("add", arguments={"a": 1, "b": 2})

            # Resource operations
            resources = await session.list_resources()
            templates = await session.list_resource_templates()
            content = await session.read_resource("config://settings")

            # Subscribe/unsubscribe to resource updates
            await session.subscribe_resource("config://settings")
            await session.unsubscribe_resource("config://settings")

            # Prompt operations
            prompts = await session.list_prompts()
            prompt_result = await session.get_prompt("greet", arguments={"name": "Alice"})

            # Completion suggestions
            completion = await session.complete(
                ref=types.PromptReference(type="ref/prompt", name="greet"),
                argument={"name": "style", "value": "for"},
                context_arguments={"name": "Alice"}
            )
            print(f"Completions: {completion.completion.values}")

            # Set logging level
            await session.set_logging_level(types.LoggingLevel("debug"))

            # Send progress notification
            await session.send_progress_notification(
                progress_token="task-123",
                progress=50.0,
                total=100.0,
                message="Processing..."
            )

            # Notify about roots list changes
            await session.send_roots_list_changed()

if __name__ == "__main__":
    asyncio.run(demonstrate_session_methods())
```

## Summary

The MCP Python SDK provides a complete implementation of the Model Context Protocol, enabling developers to build servers that expose tools, resources, and prompts to LLM applications. The high-level `MCPServer` class offers a decorator-based approach ideal for rapid development, while the low-level `Server` class provides fine-grained control for advanced use cases. Key features include multiple transport options (stdio, SSE, Streamable HTTP), OAuth 2.1 authentication support, structured output validation with Pydantic, and comprehensive context management for logging, progress reporting, and user interaction.

Common integration patterns include: deploying standalone MCP servers using the `run()` method with various transports, mounting MCP servers into existing Starlette/FastAPI applications for microservice architectures, implementing client applications using `ClientSession` for both stdio and HTTP transports, and building authenticated APIs with token verification. The SDK's async-first design ensures excellent performance for I/O-bound operations, while its type-safe context system provides clean access to MCP capabilities within handler functions. For production deployments, the Streamable HTTP transport with stateless mode and JSON responses is recommended for optimal scalability across multiple server instances.