Related Pages

Related topics: FastMCP Server Development, Project Structure

Getting Started with MCP Python SDK

Overview

The Model Context Protocol (MCP) Python SDK provides a comprehensive framework for building servers and clients that implement the MCP specification. MCP enables AI models to interact with external tools, resources, and prompts through a standardized protocol.

This guide walks you through setting up your development environment, understanding core concepts, and building your first MCP server.

Prerequisites

Before getting started, ensure you have the following installed:

Sources: CONTRIBUTING.md:3

Installation

Installing uv

If you don't have uv installed, follow the official installation guide:

# On Unix/macOS
curl -LsSf https://astral.sh/uv/install.sh | sh

# Or via pip
pip install uv

Setting Up the Project

1. Fork and clone the repository:

git clone https://github.com/YOUR-USERNAME/python-sdk.git
cd python-sdk

1. Install dependencies with all extras and dev tools:

uv sync --frozen --all-extras --dev

The --frozen flag ensures uv.lock is respected, preventing unintended dependency changes.

1. Set up pre-commit hooks:

uv tool install pre-commit --with pre-commit-uv --force-reinstall
pre-commit install

Sources: CONTRIBUTING.md:5-16

Core Concepts

Understanding MCP requires familiarity with three fundamental building blocks:

Server Architecture

MCP servers expose capabilities through three primary interfaces:

graph TD
    A[MCP Server] --> B[Tools]
    A --> C[Resources]
    A --> D[Prompts]
    
    B --> B1[Executable Functions]
    C --> C1[Readable Data]
    D --> D1[Templated Messages]

Sources: src/mcp/server/mcpserver/server.py:1-50

Tools

Tools are async functions that AI models can invoke. They support:

• Icons: Visual indicators for tool representation
• Annotations: Metadata about tool behavior (destructive, idempotent, etc.)
• Typed parameters: Automatic validation via Pydantic

from mcp.server import Server
from mcp.types import Icon

app = Server("my-server")

@app.tool(
    title="Get Weather",
    description="Get current weather for a location",
    icons=[
        Icon(src="data:image/png;base64,...", mime_type="image/png", sizes=["16x16", "32x32"])
    ]
)
async def get_weather(location: str, units: str = "celsius") -> str:
    """Fetch weather data for the given location."""
    return f"Weather in {location}: 22°C"

Sources: examples/mcpserver/icons_demo.py:1-30

Resources

Resources provide read-only data access. They follow a URI-based naming convention:

classDiagram
    class Resource {
        +str uri
        +str name
        +str mime_type
        +read() str | bytes
    }
    
    class FileResource {
        +Path path
        +bool is_binary
    }
    
    class FunctionResource {
        +Callable fn
    }
    
    Resource <|-- FileResource
    Resource <|-- FunctionResource

Sources: src/mcp/server/mcpserver/resources/types.py:1-80 Sources: src/mcp/server/mcpserver/resources/base.py:1-40

Prompts

Prompts are pre-defined message templates that can accept arguments:

from mcp import types

prompts = [
    types.Prompt(
        name="simple",
        description="A simple prompt with context and topic",
        arguments=[
            types.PromptArgument(
                name="context",
                description="Additional context to consider",
                required=False,
            ),
            types.PromptArgument(
                name="topic",
                description="Specific topic to focus on",
                required=True,
            ),
        ],
    )
]

Quick Start Guide

Creating Your First Server

The fastest way to create an MCP server is using the @mcp.tool() decorator:

# server.py
from mcp.server import Server

app = Server("my-first-server")

@app.tool(name="greet", description="Greet someone by name")
async def greet(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    app.run()

Sources: examples/snippets/servers/fastmcp_quickstart.py:1-15

Running the Server

MCP supports multiple transport mechanisms:

import click

@click.command()
@click.option("--port", default=8000, help="Port for HTTP transport")
@click.option(
    "--transport",
    type=click.Choice(["stdio", "streamable-http"]),
    default="stdio",
)
def main(port: int, transport: str) -> int:
    app = Server("my-server")
    
    if transport == "streamable-http":
        import uvicorn
        uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)
    else:
        from mcp.server.stdio import stdio_server
        async def run():
            async with stdio_server() as streams:
                await app.run(streams[0], streams[1], app.create_initialization_options())
        anyio.run(run)
    return 0

Sources: examples/servers/simple-prompt/server.py:50-75

Server Implementation Patterns

Handling Tools

from mcp.server import Server
from mcp.types import CallToolResult, TextContent

app = Server("my-server")

async def handle_call_tool(
    name: str, 
    arguments: dict[str, Any] | None
) -> CallToolResult:
    if name == "my_tool":
        result = await my_tool_function(**(arguments or {}))
        return CallToolResult(content=[TextContent(type="text", text=result)])
    raise ValueError(f"Unknown tool: {name}")

Handling Resources

async def handle_list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="file:///data/config.json",
            name="config",
            description="Application configuration",
            mimeType="application/json",
        )
    ]

async def handle_read_resource(uri: AnyUrl) -> Iterable[ReadResourceContents]:
    # Load and return resource content
    content = load_resource(uri)
    return [ReadResourceContents(content=content, mime_type="text/plain")]

Handling Prompts

async def handle_list_prompts() -> list[types.Prompt]:
    return [
        types.Prompt(
            name="summarize",
            description="Summarize a document",
            arguments=[
                types.PromptArgument(
                    name="text",
                    description="Text to summarize",
                    required=True,
                )
            ],
        )
    ]

async def handle_get_prompt(
    name: str, 
    arguments: dict[str, Any] | None
) -> types.GetPromptResult:
    if name != "summarize":
        raise ValueError(f"Unknown prompt: {name}")
    
    return types.GetPromptResult(
        messages=[{"role": "user", "content": {"type": "text", "text": f"Summarize: {arguments['text']}"}}],
        description="Document summarization prompt",
    )

Advanced Features

Resource Templates

Resource templates allow dynamic resource generation based on URI patterns:

from mcp.server import Server

app = Server("github-server")

@app.list_resource_templates()
async def list_templates() -> list[types.ResourceTemplate]:
    return [
        types.ResourceTemplate(
            uri_template="github://repos/{owner}/{repo}",
            name="repository",
            description="Access a GitHub repository",
            arguments=[
                types.PromptArgument(name="owner", required=True),
                types.PromptArgument(name="repo", required=True),
            ],
        )
    ]

Tool Annotations

Annotations provide metadata about tool behavior:

from mcp.types import Annotations

@app.tool(
    annotations=Annotations(
        destructive=AnnotationState.UNSPECIFIED,
        idempotent=AnnotationState.TRUE,
        side_effects=AnnotationState.FALSE,
    )
)
async def safe_operation(data: str) -> str:
    return data.upper()

Client Configuration

MCP clients need to know how to connect to servers. Configuration varies by client:

Environment Variables

Claude Desktop Integration

To install a server in Claude Desktop:

mcp install examples/servers/my-server/server.py \
    --name "My Server" \
    --with-editable ./ \
    --env-var API_KEY=secret123

The server will be available in Claude Desktop's MCP configuration.

Development Workflow

Code Quality Standards

Running Tests

# All tests
uv run pytest

# Specific Python version
uv run --frozen --python 3.10 pytest

# With coverage
uv run pytest --cov=mcp --cov-report=term-missing

Pre-commit Hooks

Pre-commit runs all quality checks automatically:

# Install hooks
pre-commit install

# Run on all files
pre-commit run --all-files

# Run specific hook
pre-commit run ruff --all-files

Sources: CONTRIBUTING.md:30-45

Exception Handling Best Practices

Follow these guidelines for robust error handling:

Always use logger.exception() when catching exceptions:

try:
    resource = await resource_manager.get_resource(uri)
except Exception as exc:
    logger.exception(f"Error getting resource {uri}")
    raise ResourceError(f"Error reading resource {uri}") from exc

Sources: CONTRIBUTING.md:55-65

Next Steps

After completing this guide, explore:

1. Example Servers: Browse examples/servers/ for complete implementations

• simple-prompt/ - Prompt handling
• simple-pagination/ - Pagination patterns
• simple-task-interactive/ - Interactive tasks with elicitation

1. API Reference: Check src/mcp/ for the full SDK implementation

1. Specification: Review the MCP protocol specification for protocol details

1. Contributing: See CONTRIBUTING.md for contribution guidelines and coding standards

Quick Reference

Minimal Server Template

from mcp.server import Server
from mcp import types

app = Server("my-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [types.Tool(name="hello", description="Say hello", inputSchema={})]

@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.Content]:
    if name == "hello":
        return [types.TextContent(type="text", text="Hello!")]
    raise ValueError(f"Unknown tool: {name}")

if __name__ == "__main__":
    app.run()

Installation Checklist

• [ ] Python 3.10+ installed
• [ ] uv installed
• [ ] Repository cloned
• [ ] Dependencies installed (uv sync --frozen --all-extras --dev)
• [ ] Pre-commit hooks configured
• [ ] First server implemented
• [ ] Code formatted and linted

Related Pages

Related topics: Getting Started with MCP Python SDK, FastMCP Server Development

Project Structure

The Model Context Protocol (MCP) Python SDK is organized as a comprehensive toolkit for building both MCP clients and servers. The project structure reflects the protocol's architecture, separating concerns between client-side, server-side, and shared components.

Repository Layout

The repository follows a standard Python package layout with additional documentation and examples:

Source Code Organization

Core Package (`src/mcp/`)

The main package contains several subpackages that separate the SDK into logical layers:

graph TD
    A[src/mcp/] --> B[client/]
    A --> C[server/]
    A --> D[shared/]
    A --> E[types/]
    C --> F[mcpserver/]
    C --> G[auth/]
    F --> H[resources/]

Client Package (`src/mcp/client/`)

The client package provides functionality for connecting to MCP servers. Clients handle the protocol-level communication, including transport management and request/response handling.

Server Package (`src/mcp/server/`)

The server package is the core of the SDK, containing:

Shared Package (`src/mcp/shared/`)

Shared utilities used by both clients and servers, including common data structures and helper functions.

Types Package (`src/mcp/types/`)

Type definitions for MCP protocol entities such as resources, prompts, tools, and annotations.

MCPServer Architecture

The MCPServer class is the primary entry point for building MCP servers. It wraps a low-level Server and provides high-level abstractions for registering resources, prompts, and tools.

graph TD
    A[MCPServer] --> B[ResourceManager]
    A --> C[PromptManager]
    A --> D[ToolHandler]
    B --> E[FunctionResource]
    B --> F[FileResource]
    B --> G[ResourceTemplate]

Resource Management

Resources are managed through the ResourceManager class, which handles both static resources and dynamic resource templates. The base class Resource defines the common interface:

• uri: Unique identifier for the resource
• name: Optional name for the resource
• title: Human-readable title
• description: Description of the resource
• mime_type: MIME type of the content (default: text/plain)
• icons: Optional list of icons
• annotations: Optional annotations
• meta: Optional metadata dictionary

Sources: src/mcp/server/mcpserver/resources/base.py:1-50

Resource Types

Function resources are created from functions using the from_function() classmethod:

@staticmethod
def from_function(
    fn: Callable[..., Any],
    uri: str,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
) -> FunctionResource

Sources: src/mcp/server/mcpserver/resources/types.py:1-80

Tool Registration

Tools are registered using the add_tool() decorator or method. The server handles the routing of tool calls through the _handle_call_tool callback:

def add_tool(
    self,
    fn: Callable[..., Any],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: Annotations | None = None,
) -> Callable[[_CallableT], _CallableT]

Sources: src/mcp/server/mcpserver/server.py:1-200

Prompt Management

Prompts are registered using the prompt() decorator and managed by the PromptManager. The get_prompt() method renders prompts with arguments:

async def get_prompt(
    self, 
    name: str, 
    arguments: dict[str, Any] | None = None,
    context: Context[LifespanResultT, Any] | None = None
) -> GetPromptResult

Sources: src/mcp/server/mcpserver/server.py:200-300

Authentication Architecture

The auth package implements OAuth 2.0 support for MCP servers. Key components include:

Metadata endpoints follow RFC 9728 compliance, constructing URLs by inserting /.well-known/oauth-protected-resource between the host and resource path.

Sources: src/mcp/server/auth/routes.py:1-100

Transport Layer

MCP servers support multiple transport mechanisms:

The server creates different application instances based on the transport:

if transport == "streamable-http":
    uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)
else:
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

Sources: examples/servers/simple-prompt/mcp_simple_prompt/server.py:1-80

Examples Structure

The examples/ directory contains working implementations:

examples/
├── servers/
│   ├── simple-prompt/
│   ├── simple-task-interactive/
│   └── ...
└── clients/
    ├── simple-chatbot/
    ├── simple-task-client/
    └── ...

Server Examples

Server examples demonstrate:

• Tool registration patterns
• Resource and prompt management
• Interactive tasks with elicitation
• Authentication integration

Client Examples

Client examples demonstrate:

• Connecting via different transports
• Calling tools and resources
• Polling for task results
• OAuth authentication flows

Key Files Reference

Development Guidelines

The project enforces strict code quality standards:

• Type hints required for all public APIs
• Docstrings required for public APIs with Raises: sections for documented exceptions
• Exception handling: Use logger.exception() instead of logger.error() when catching exceptions
• Testing: Use pytest with anyio for async testing

Sources: AGENTS.md

Summary

The MCP Python SDK is structured around a clean separation between:

1. Protocol types (types/) - Definitions of MCP entities
2. Server implementation (server/) - Building MCP servers with resources, prompts, and tools
3. Client implementation (client/) - Connecting to and interacting with servers
4. Shared utilities (shared/) - Common functionality

The architecture supports multiple transports and authentication mechanisms while maintaining a consistent interface through the MCPServer class.

Related Pages

Related topics: Low-Level Server Development, Server Lifecycle and Context Management, Resources, Tools and Structured Output, Prompts and Templates

FastMCP Server Development

FastMCP is a streamlined framework for building Model Context Protocol (MCP) servers in Python. It provides a simplified API over the lower-level Server class, making it easier to expose tools, resources, prompts, and other MCP capabilities to LLM clients.

Architecture Overview

FastMCP follows a modular architecture where different components handle distinct aspects of the MCP protocol. The framework is built around a central server instance that orchestrates tool execution, resource management, and prompt handling.

The MCP server architecture consists of several key layers working together to provide a complete protocol implementation. At the lowest level, the Server class handles protocol-level concerns including message parsing, state management, and transport abstraction. Above this, FastMCP provides a more developer-friendly interface for registering handlers and exposing capabilities.

graph TD
    A[Client] --> B[Transport Layer<br/>stdio or HTTP]
    B --> C[FastMCP Server]
    C --> D[Tools Handler]
    C --> E[Resources Handler]
    C --> F[Prompts Handler]
    D --> G[Business Logic]
    E --> H[Resource Storage]
    F --> I[Prompt Templates]

Core Components

Server Initialization

FastMCP servers are initialized with a name and optional configuration parameters. The server instance serves as the central registry for all tools, resources, and prompts that will be exposed to clients.

The following table outlines the key configuration options available when creating a FastMCP server:

Sources: src/mcp/server/mcpserver/server.py:80-120

Tool Registration

Tools are the primary mechanism for servers to expose executable functionality to clients. FastMCP provides the add_tool method for registering callable functions with automatic schema generation.

@mcp.add_tool()
def get_time() -> dict:
    """Get the current server time."""
    return {"current_time": "2024-01-15T10:30:00", "timezone": "UTC"}

Tools can be registered with custom names, titles, and descriptions to control how they appear to clients. The decorator-based approach automatically extracts docstrings and type hints to generate the tool's JSON schema.

Resource Management

Resources provide a way to expose data that clients can read. FastMCP supports both static resources and dynamic templates that can be resolved at read time.

Sources: src/mcp/server/mcpserver/server.py:40-60

Prompt Templates

Prompts allow servers to expose reusable prompt templates that clients can invoke with arguments. This enables servers to provide standardized interactions for common tasks.

Transport Configuration

FastMCP supports multiple transport mechanisms for client-server communication. The choice of transport affects how clients connect and interact with the server.

Stdio Transport

The stdio transport uses standard input and output streams for message passing. This is the default transport and works well for local integrations.

from mcp.server.stdio import stdio_server

async def arun():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

anyio.run(arun)

Sources: examples/servers/simple-pagination/mcp_simple_pagination/server.py:80-90

Streamable HTTP Transport

For networked deployments, the streamable HTTP transport provides persistent connections over HTTP. This enables longer-running operations and streaming responses.

Sources: examples/clients/simple-auth-client/README.md:50-60

Task Execution Model

FastMCP supports an experimental task execution model that allows tools to be called as asynchronous tasks. This enables longer-running operations with progress tracking and intermediate status updates.

The task lifecycle follows a specific state progression:

stateDiagram-v2
    [*] --> Created: task created
    Created --> Working: execution started
    Working --> Working: progress update
    Working --> Completed: success
    Working --> Failed: error
    Completed --> [*]
    Failed --> [*]

Tasks are initiated using the experimental call_tool_as_task method, which returns immediately with a task reference. Clients then poll for status updates using get_task_result.

Sources: examples/clients/simple-task-client/README.md:25-35

Interactive Task Features

FastMCP supports two advanced interaction patterns for tasks that require external input during execution.

Elicitation

Elicitation allows a task to pause and request user confirmation before continuing. The server uses task.elicit() to send a prompt to the client, which responds with the user's input.

async def confirm_delete(filename: str, task=None):
    if task:
        result = await task.elicit(
            message="Confirm deletion",
            params=[...]
        )
        if result.action == "accept":
            # proceed with deletion
            pass

Sources: examples/servers/simple-task-interactive/README.md:15-25

Sampling

Sampling enables tasks to request LLM completions during execution. The server sends a prompt and sampling parameters to the client, which returns the generated text.

async def write_haiku(topic: str, task=None):
    if task:
        response = await task.create_message(
            messages=[...],
            model_preferences={...}
        )
        return response.content

Dependency Management

FastMCP uses uv for package management in all development and deployment scenarios. The project maintains strict dependency floors and uses frozen lock files for reproducible environments.

Sources: CONTRIBUTING.md:10-20

Code Quality Standards

FastMCP enforces specific quality standards for all contributions:

• Type hints: Required for all public APIs
• Docstrings: Required for public functions and classes
• Linting: ruff check . --fix
• Formatting: ruff format .
• Type checking: pyright

The project uses strict-no-cover to ensure test coverage for all non-trivial code paths. New code should include appropriate test coverage before submission.

Sources: AGENTS.md:30-50

Exception Handling Patterns

FastMCP defines specific exception handling patterns that all server implementations should follow:

Sources: AGENTS.md:25-30

Testing Guidelines

Tests for FastMCP components should follow these principles:

• Use anyio for async testing, not raw asyncio
• Write plain top-level test_* functions, not Test prefixed classes
• Tests should be fast and deterministic
• Prefer in-memory execution over subprocess spawning

uv run --frozen pytest

Sources: AGENTS.md:55-65

Complete Server Example

The following example demonstrates a minimal FastMCP server with tools, resources, and prompts:

import anyio
import click
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("example-server")

@mcp.add_tool()
def get_time() -> dict:
    """Get the current server time."""
    return {"current_time": "2024-01-15T10:30:00", "timezone": "UTC"}

@mcp.add_resource("file://example/{name}")
def get_file(name: str) -> str:
    return f"Content of {name}"

@mcp.add_prompt()
def review_code(code: str, language: str) -> str:
    return f"Review this {language} code:\n\n{code}"

@click.command()
@click.option("--port", default=8000)
@click.option("--transport", default="stdio")
def main(port: int, transport: str) -> int:
    if transport == "streamable-http":
        import uvicorn
        uvicorn.run(mcp.streamable_http_app(), host="127.0.0.1", port=port)
    else:
        anyio.run(mcp.run)
    return 0

Next Steps

For continued learning about FastMCP development:

1. Explore the example servers directory for complete implementations
2. Review the migration guide for breaking changes between versions
3. Consult the MCP specification for protocol-level details

Related Pages

Related topics: FastMCP Server Development, Server Lifecycle and Context Management, Transports Overview

Low-Level Server Development

The Model Context Protocol (MCP) Python SDK provides a low-level server API that gives developers fine-grained control over protocol handling, session management, and request/response processing. This approach differs from the high-level MCPServer abstraction by exposing the underlying protocol mechanics directly, enabling custom transport implementations, advanced middleware, and specialized integration scenarios.

Overview

The low-level server layer is the foundational component of the MCP SDK's server architecture. It operates at the protocol level, handling JSON-RPC message processing, session lifecycle, and notification dispatching without imposing opinions on transport mechanisms or application structure.

Key Characteristics

The low-level server is used internally by the higher-level MCPServer class, which adds convenience abstractions, automatic middleware, and resource/prompt management. Direct use of the low-level API is appropriate when:

• Building custom MCP server implementations
• Implementing specialized transport adapters
• Requiring direct control over protocol behavior
• Developing protocol-level tools or testing utilities

Architecture

Component Hierarchy

graph TD
    A[Low-Level Server] --> B[Session Manager]
    A --> C[Request Handler]
    A --> D[Notification Dispatcher]
    B --> E[Client Session]
    B --> F[Client Session]
    E --> G[Protocol State]
    F --> H[Protocol State]
    C --> I[List Tools Handler]
    C --> J[Call Tool Handler]
    C --> K[List Resources Handler]
    C --> L[Read Resource Handler]

Server Class Structure

The Server class in src/mcp/server/lowlevel/server.py serves as the central coordinator for all protocol operations. It accepts callback handlers for each protocol method and manages the registration of server capabilities.

class Server:
    def __init__(
        self,
        name: str,
        title: str | None = None,
        description: str | None = None,
        instructions: str | None = None,
        version: str | None = None,
        on_list_tools: Callable | None = None,
        on_call_tool: Callable | None = None,
        on_list_resources: Callable | None = None,
        on_read_resource: Callable | None = None,
        on_list_resource_templates: Callable | None = None,
        on_list_prompts: Callable | None = None,
        on_get_prompt: Callable | None = None,
        on_list_resource_subscriptions: Callable | None = None,
        on_subscribe_resource: Callable | None = None,
        on_unsubscribe_resource: Callable | None = None,
        on_list_completions: Callable | None = None,
        on_initialize: Callable | None = None,
        on_set_logging_level: Callable | None = None,
        on_send_notification: Callable | None = None,
        lifespan: Callable | None = None,
    )

Request Handlers

The low-level server uses callback-based handlers for each protocol method. These handlers receive structured request parameters and return typed responses.

Tool Handlers

Tools are executable functions exposed by the server to clients. The low-level API provides two handlers:

Resource Handlers

Resources provide read-only data access through URIs:

Prompt Handlers

Prompts are templated message configurations:

Completion Handler

Auto-completion support for resource template variables:

on_list_completions: Callable[[ServerRequestContext, types.CompleteRequestParams], Awaitable[types.CompleteResult]]

Lifecycle Handlers

Session Management

The session layer (src/mcp/server/session.py) manages individual client connections. Each session maintains protocol state, request/response tracking, and notification channels.

Session Lifecycle

stateDiagram-v2
    [*] --> Initializing: Client connects
    Initializing --> Running: Initialize response sent
    Running --> Running: Request/Response cycle
    Running --> Closed: Client disconnects
    Closed --> [*]: Cleanup complete

Session Responsibilities

• Protocol State: Maintains initialization state and capability negotiation
• Request Tracking: Correlates requests with responses using JSON-RPC IDs
• Notification Queue: Buffers notifications for delivery
• Resource Subscriptions: Tracks active resource subscriptions per client

Lifespan Management

Lifespan handlers manage server startup and shutdown phases, enabling resource acquisition and cleanup.

Lifespan Callback Pattern

from contextlib import asynccontextmanager
from mcp.server.lowlevel.server import Server

@asynccontextmanager
async def lifespan(server: Server):
    # Startup: acquire resources
    db = await connect_to_database()
    cache = await create_cache()
    
    # Make resources available to handlers
    server.request_context.append(db)
    
    yield
    
    # Shutdown: release resources
    await db.close()
    await cache.close()

Lifespan Phases

Sources: examples/snippets/servers/lowlevel/lifespan.py

Basic Implementation Pattern

A minimal low-level server implementation follows this pattern:

from mcp.server.lowlevel.server import Server
from mcp.types import Tool, CallToolResult

async def handle_list_tools() -> list[Tool]:
    return [
        Tool(
            name="hello",
            description="Say hello to someone",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string"}
                }
            }
        )
    ]

async def handle_call_tool(
    name: str, arguments: dict | None
) -> CallToolResult:
    if name == "hello":
        return CallToolResult(
            content=[types.TextContent(type="text", text=f"Hello, {arguments.get('name', 'World')}!")]
        )
    raise ValueError(f"Unknown tool: {name}")

server = Server(
    name="hello-server",
    on_list_tools=handle_list_tools,
    on_call_tool=handle_call_tool,
)

Sources: examples/snippets/servers/lowlevel/basic.py

Integration with Transports

The low-level server is transport-agnostic. Different transport implementations connect to the server:

Stdio Transport

For command-line integration and local processes:

from mcp.server.stdio import stdio_server

async def run():
    async with stdio_server() as streams:
        await server.run(
            streams[0],
            streams[1],
            server.create_initialization_options()
        )

Sources: examples/servers/simple-prompt/mcp_simple_prompt/server.py

Streamable HTTP Transport

For networked deployments with SSE notifications:

import uvicorn

app = server.streamable_http_app()
uvicorn.run(app, host="127.0.0.1", port=8000)

Custom Transport Integration

Implement custom transports by calling server methods directly:

# Custom transport reads from WebSocket
async def custom_transport_handler(websocket):
    read_stream = websocket_async_reader(websocket)
    write_stream = websocket_async_writer(websocket)
    
    await server.run(
        read_stream,
        write_stream,
        server.create_initialization_options()
    )

Server Initialization Options

The create_initialization_options() method generates the protocol initialization payload:

def create_initialization_options(
    self,
    capabilities: ServerCapabilities | None = None,
    protocol_version: str | None = None,
) -> InitializationOptions

Capabilities are automatically constructed based on registered handlers, but can be explicitly overridden.

Error Handling

The low-level server expects handlers to raise appropriate exceptions for error conditions:

Handlers should use logger.exception() for logging errors rather than logger.error() when catching exceptions, ensuring stack traces are captured for debugging.

Sources: AGENTS.md

Testing Low-Level Servers

The SDK uses pytest with anyio for async testing. Test patterns for low-level servers:

import pytest
from mcp.server.lowlevel.server import Server
from mcp.testing import create_test_session

async def test_tool_discovery():
    server = Server(name="test", on_list_tools=list_tools_handler)
    
    async with create_test_session(server) as session:
        tools = await session.list_tools()
        assert len(tools) == 1
        assert tools[0].name == "test_tool"

Sources: CONTRIBUTING.md

High-Level vs Low-Level API

The MCPServer class internally delegates to the low-level Server, adding convenience features. For most use cases, MCPServer provides sufficient functionality with less boilerplate.

Sources: src/mcp/server/mcpserver/server.py

Best Practices

1. Always use logger.exception() when logging caught exceptions
2. Catch specific exceptions rather than broad except Exception:
3. Use type hints on all handler functions and public APIs
4. Include docstrings for handler functions with Raises: sections
5. Test handlers in isolation using mock sessions
6. Validate handler arguments before processing
7. Clean up resources in lifespan shutdown phases

Public API Surface

The public API is defined in src/mcp/server/lowlevel/__init__.py via __all__. Only symbols listed there should be imported:

from mcp.server.lowlevel import (
    Server,
    ServerRequestContext,
    InitializationOptions,
)

Adding symbols to __all__ is a deliberate API decision, not a convenience re-export.

Sources: AGENTS.md

Related Pages

Related topics: FastMCP Server Development, Low-Level Server Development, Context and Session Management

Server Lifecycle and Context Management

The Model Context Protocol (MCP) Python SDK implements a sophisticated lifecycle and context management system that enables servers to handle initialization, request processing, resource access, and graceful shutdown. This documentation covers the architecture, components, and best practices for implementing servers that properly manage their lifecycle and context propagation.

Overview

The MCP server architecture is built around three core concepts:

1. Lifecycle Management: Managing server startup, request handling, and graceful shutdown through lifespan handlers
2. Context Objects: Request-scoped objects that carry server references, arguments, and metadata through handler chains
3. Resource Managers: Components that handle resource registration, retrieval, and templating with contextual awareness

These components work together to provide a consistent, testable, and maintainable server implementation. The MCPServer class serves as the central coordinator, integrating the low-level Server with resource managers, prompt managers, and lifespan handling.

Architecture

Component Hierarchy

graph TD
    A[MCP Server Application] --> B[MCPServer]
    B --> C[LowLevel Server]
    C --> D[Lifespan Manager]
    B --> E[Resource Manager]
    B --> F[Prompt Manager]
    B --> G[Context Factory]
    
    H[Client Request] --> C
    C --> I[Request Handlers]
    I --> G
    G --> E
    G --> F

Lifecycle Flow

sequenceDiagram
    participant Client
    participant MCPServer
    participant LifespanManager
    participant ResourceManager
    participant PromptManager
    
    Note over MCPServer: Startup Phase
    MCPServer->>LifespanManager: Initialize lifespan context
    LifespanManager->>UserLifespan: startup callback
    UserLifespan-->>LifespanManager: lifespan_state
    
    Note over MCPServer: Ready State
    Client->>MCPServer: list_tools request
    MCPServer->>ResourceManager: Get registered tools
    ResourceManager-->>MCPServer: Tool list
    
    Client->>MCPServer: read_resource request
    MCPServer->>LifespanManager: Get context
    LifespanManager-->>MCPServer: context with lifespan_state
    MCPServer->>ResourceManager: Get resource with context
    ResourceManager-->>MCPServer: Resource content
    
    Note over MCPServer: Shutdown Phase
    MCPServer->>LifespanManager: Shutdown signal
    LifespanManager->>UserLifespan: shutdown callback

Lifecycle Management

Lifespan Wrapper

The lifespan_wrapper function wraps user-defined lifespan callbacks and integrates them with the server's low-level implementation:

lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan)

Sources: src/mcp/server/mcpserver/server.py:100-101

The lifespan wrapper serves two primary purposes:

1. Initialization: Creates the lifespan context that will be available throughout the server's lifetime
2. Integration: Bridges user-defined startup/shutdown callbacks with the low-level server's lifespan protocol

Default Lifespan

When no custom lifespan is provided, the default_lifespan ensures the server still follows proper lifecycle protocols. This is particularly useful for simple servers that don't require initialization or cleanup logic.

Lifespan Context State

The lifespan context (LifespanResultT) carries server-wide state that handlers can access. This state is created during startup and remains available until shutdown:

async def read_resource(
    self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
) -> Iterable[ReadResourceContents]:
    """Read a resource by URI."""
    if context is None:
        context = Context(mcp_server=self)

Sources: src/mcp/server/mcpserver/server.py:47-51

The context parameter is optional but recommended for accessing lifespan state within resource handlers.

Example: Custom Lifespan Implementation

from contextlib import asynccontextmanager
from mcp.server import MCPServer

@asynccontextmanager
async def my_lifespan(server: MCPServer):
    # Startup: Initialize connections, load data
    db_connection = await connect_to_database()
    server.context["db"] = db_connection
    
    yield  # Server runs here
    
    # Shutdown: Clean up resources
    await db_connection.close()

app = MCPServer(
    name="my-server",
    lifespan=my_lifespan
)

Context Management

Context Object Structure

The Context object is a generic container that provides access to server state and request metadata:

The context is parameterized with two type variables:

Context[LifespanResultT, Any]

Where LifespanResultT is the type of the lifespan state, and Any represents additional request-scoped data.

Context in Resource Operations

When reading resources, the context carries critical information that resource handlers can use:

async def read_resource(
    self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
) -> Iterable[ReadResourceContents]:
    if context is None:
        context = Context(mcp_server=self)
    try:
        resource = await self._resource_manager.get_resource(uri, context)

Sources: src/mcp/server/mcpserver/server.py:47-56

The resource manager passes this context to resource handlers, enabling:

• Dynamic content: Resources can read lifespan state (e.g., database connections) to generate content
• Template resolution: URI templates can use context arguments to resolve placeholders
• Metadata propagation: Request metadata flows through to resource handlers

Context in Completion Resolution

The context object also supports completion callbacks, where it carries the arguments from the current request:

if context and context.arguments and context.arguments.get("owner") == "modelcontextprotocol":
    repos = ["python-sdk", "typescript-sdk", "specification"]
    return Completion(values=repos, has_more=False)

Sources: examples/snippets/servers/completion.py:8-13

Resource Management

Resource Manager Initialization

The ResourceManager handles all resource-related operations:

self._resource_manager = ResourceManager(
    resources=resources, 
    warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources
)

Sources: src/mcp/server/mcpserver/server.py:85-87

Resource Types

The SDK provides several resource types for different use cases:

FunctionResource with Context

The FunctionResource class accepts a callable that receives context for dynamic content generation:

@classmethod
def from_function(
    cls,
    fn: Callable[..., Any],
    uri: str | None = None,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: ResourceIcons | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
) -> FunctionResource:
    """Create a FunctionResource from a function."""
    func_name = name or fn.__name__
    if func_name == "<lambda>":  # pragma: no cover
        raise ValueError("You must provide a name for lambda functions")

    # ensure the arguments are properly cast
    fn = validate_call(fn)

    return cls(
        uri=uri,
        name=func_name,
        title=title,
        description=description or fn.__doc__ or "",
        mime_type=mime_type or "text/plain",
        fn=fn,
        icons=icons,
        annotations=annotations,
        meta=meta,
    )

Sources: src/mcp/server/mcpserver/resources/types.py:101-130

The function is validated using validate_call to ensure proper argument handling and type safety.

FileResource with Binary Support

The FileResource handles file-based resources with explicit binary support:

class FileResource(Resource):
    """A resource that reads from a file.

    Set is_binary=True to read the file as binary data instead of text.
    """

    path: Path = Field(description="Path to the file")
    is_binary: bool = Field(
        default=False,
        description="Whether to read the file as binary data",
    )
    mime_type: str = Field(
        default="text/plain",
        description="MIME type of the resource content",
    )

    @pydantic.field_validator("path")
    @classmethod
    def validate_absolute_path(cls, path: Path) -> Path:
        """Ensure path is absolute."""
        if not path.is_absolute():
            raise ValueError("Path must be absolute")
        return path

Sources: src/mcp/server/mcpserver/resources/types.py:133-154

Key validation rules:

• Absolute paths required: All file paths must be absolute to prevent path traversal issues
• MIME type inference: Can be manually specified or derived from file extension
• Binary mode: Set is_binary=True for non-text content

Request Handling Architecture

Handler Chain

graph LR
    A[Incoming Request] --> B[MCPServer Entry]
    B --> C[List Tools Handler]
    B --> D[Call Tool Handler]
    B --> E[List Resources Handler]
    B --> F[Read Resource Handler]
    B --> G[List Prompts Handler]
    B --> H[Get Prompt Handler]
    
    C --> I[Resource Manager]
    E --> I
    F --> I
    I --> J[Context Injection]
    J --> K[Resource Handler]

Error Handling in Resource Reading

The server implements defensive error handling to prevent information leakage:

try:
    resource = await self._resource_manager.get_resource(uri, context)
except ValueError as exc:
    raise ResourceError(f"Unknown resource: {uri}") from exc

try:
    content = await resource.read()
    return [ReadResourceContents(content=content, mime_type=resource.mime_type, meta=resource.meta)]
except Exception as exc:
    logger.exception(f"Error getting resource {uri}")
    # If an exception happens when reading the resource, we should not leak the exception to the client.
    raise ResourceError(f"Error reading resource {uri}") from exc

Sources: src/mcp/server/mcpserver/server.py:53-67

Key principles:

1. Specific exception catching: ValueError is caught for unknown resources
2. Generic exception wrapping: Unexpected errors are wrapped to prevent leaking implementation details
3. Logging with context: logger.exception() is used to preserve stack traces for debugging

Prompt and Tool Integration

Prompt Manager

The PromptManager handles prompt registration with duplicate detection:

self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)

Sources: src/mcp/server/mcpserver/server.py:88-89

Handler Registration

All handlers are registered with the low-level server during initialization:

self._lowlevel_server = Server(
    name=name or "mcp-server",
    title=title,
    description=description,
    instructions=instructions,
    website_url=website_url,
    icons=icons,
    version=version,
    on_list_tools=self._handle_list_tools,
    on_call_tool=self._handle_call_tool,
    on_list_resources=self._handle_list_resources,
    on_read_resource=self._handle_read_resource,
    on_list_resource_templates=self._handle_list_resource_templates,
    on_list_prompts=self._handle_list_prompts,
    on_get_prompt=self._handle_get_prompt,
    lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),
)

Sources: src/mcp/server/mcpserver/server.py:90-109

Transport Integration

Streamable HTTP Transport

Servers can expose HTTP endpoints for remote clients:

@click.command()
@click.option("--port", default=8000, help="Port to listen on for HTTP")
@click.option(
    "--transport",
    type=click.Choice(["stdio", "streamable-http"]),
    default="stdio",
    help="Transport type",
)
def main(port: int, transport: str) -> int:
    app = Server("mcp-simple-prompt", ...)
    
    if transport == "streamable-http":
        import uvicorn
        uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)

Sources: examples/servers/simple-prompt/mcp_simple_prompt/server.py:42-56

The streamable_http_app() method creates an ASGI application that handles the MCP protocol over HTTP, maintaining lifecycle and context across requests.

Best Practices

Lifecycle Management

Context Usage

Resource Handling

Testing Considerations

According to the development guidelines, tests should be:

• Fast and deterministic: Prefer in-memory async execution
• Use anyio for async testing: Not asyncio directly
• Avoid Test-prefixed classes: Write plain test_* functions
• Reach for threads only when necessary: Subprocesses as last resort

Sources: AGENTS.md:75-82

The context-aware design enables easy testing by allowing mock contexts to be injected:

async def test_resource_with_context():
    # Create a mock context with test lifespan state
    context = Context(
        mcp_server=test_server,
        arguments={"owner": "test-owner"}
    )
    
    # Resource handler can access context.arguments
    result = await server.read_resource("github://repos/{owner}/{repo}", context)

Summary

The MCP Python SDK's server lifecycle and context management system provides:

1. Structured lifecycle handling: Startup, request processing, and shutdown phases with proper resource management
2. Context propagation: Request-scoped objects carrying server references and arguments through handler chains
3. Resource abstraction: Flexible resource types supporting static files and dynamic content
4. Error isolation: Protection against information leakage while maintaining debuggability
5. Transport flexibility: Support for both stdio and streamable HTTP transports

This architecture enables developers to build robust MCP servers that maintain clean separation of concerns, proper resource lifecycle management, and consistent request handling patterns.

Related Pages

Related topics: FastMCP Server Development, Prompts and Templates, Tools and Structured Output

Resources

Resources are a core concept in the Model Context Protocol (MCP) that allow servers to expose data and content to clients. Unlike tools (which perform actions) or prompts (which are templates for LLM interactions), resources provide a way to share information that can be read by clients.

Overview

In MCP, resources serve as data containers that can be:

• Read on demand by clients
• Organized with metadata (title, description, MIME type)
• Annotated with additional context
• Dynamically generated (via functions) or static (from files)

Resources follow a request-response pattern where clients query for available resources and then read their contents when needed.

Sources: examples/snippets/servers/basic_resource.py:1-50

Architecture

graph TD
    A[Client] -->|list_resources / list_resource_templates| B[MCPServer]
    A -->|read_resource| B
    B -->|delegates| C[ResourceManager]
    C -->|manages| D[Resource instances]
    C -->|manages| E[ResourceTemplate instances]
    D -->|read| F[FunctionResource]
    D -->|read| G[FileResource]
    F -->|dynamic content| H[Callable Function]
    G -->|static content| I[File System]

Resource Types

Base Resource

All resources inherit from the abstract Resource base class defined in base.py:

Sources: src/mcp/server/mcpserver/resources/base.py:17-31

FunctionResource

A resource that generates content dynamically by calling a function:

class FunctionResource(Resource):
    fn: Callable[[], str | bytes]

The function is invoked each time a client reads the resource, allowing for dynamic content generation. The function should return either:

• str for text content
• bytes for binary content

Sources: src/mcp/server/mcpserver/resources/types.py:1-50

FileResource

A resource that reads content from the file system:

class FileResource(Resource):
    path: Path
    is_binary: bool = False
    mime_type: str = "text/plain"

Key validation rules:

• Path must be absolute (raises ValueError otherwise)
• Binary mode is automatically detected based on MIME type when not explicitly set

Sources: src/mcp/server/mcpserver/resources/types.py:80-100

Resource Templates

Resource templates allow servers to expose parameterized resources. Clients can list templates and provide arguments to instantiate specific resource URIs.

Template Structure

classDiagram
    class ResourceTemplate {
        +uri_template: str
        +name: str
        +title: str | None
        +description: str | None
        +mime_type: str
        +annotations: Annotations | None
        +_meta: dict
    }

Templates use URI templates with placeholders that clients fill in when making requests.

Sources: src/mcp/server/fastmcp/resources/templates.py

Resource Manager

The ResourceManager handles registration, storage, and retrieval of resources and templates:

The manager supports duplicate resource detection controlled by the warn_on_duplicate_resources setting.

Sources: src/mcp/server/fastmcp/resources/resource_manager.py

Server Integration

The MCPServer class integrates resources through the resource manager:

sequenceDiagram
    participant Client
    participant MCPServer
    participant ResourceManager
    participant Resource
    
    Client->>MCPServer: list_resources()
    MCPServer->>ResourceManager: list_resources()
    ResourceManager-->>MCPServer: List[Resource]
    MCPServer-->>Client: MCPResource list
    
    Client->>MCPServer: read_resource(uri)
    MCPServer->>ResourceManager: get_resource(uri)
    ResourceManager->>Resource: read()
    Resource-->>ResourceManager: content
    ResourceManager-->>MCPServer: ReadResourceContents
    MCPServer-->>Client: Resource content

The server implements these async methods:

• list_resources() → list[MCPResource]
• list_resource_templates() → list[MCPResourceTemplate]
• read_resource(uri) → Iterable[ReadResourceContents]

Sources: src/mcp/server/mcpserver/server.py:50-80

Usage Examples

Basic Resource Registration

from mcp.server import Server
from mcp.server.resource_manager import ResourceManager
from mcp.server.resources import FunctionResource

# Create resource manager
resource_manager = ResourceManager()

# Register a function-based resource
resource_manager.add(
    FunctionResource(
        uri="example://current-time",
        name="current_time",
        title="Current Time",
        description="Returns the current server time",
        fn=lambda: datetime.now().isoformat()
    )
)

# Create server with resources
server = Server("my-server", resources=[resource_manager])

Sources: examples/snippets/servers/basic_resource.py

Resource with Completion Support

Resources can be paired with completion providers for intelligent suggestions:

async def handle_completion(
    context: ServerRequestContext,
    ref: ResourceTemplateReference,
    argument: str
) -> Completion | None:
    if ref.uri == "github://repos/{owner}/{repo}":
        if argument.name == "repo":
            if context.arguments.get("owner") == "modelcontextprotocol":
                return Completion(
                    values=["python-sdk", "typescript-sdk", "specification"],
                    has_more=False
                )
    return None

Sources: examples/snippets/servers/completion.py:10-20

Error Handling

The server handles resource-related errors gracefully:

Internal exceptions are never leaked to clients for security reasons. The server logs the actual exception via logger.exception() and returns a sanitized error message.

Sources: src/mcp/server/mcpserver/server.py:65-75

Best Practices

1. Use descriptive URIs: Follow a consistent naming scheme like scheme://path/to/resource
2. Provide titles and descriptions: Help clients understand resource purpose
3. Set correct MIME types: Enable proper content handling by clients
4. Use function resources for dynamic content: File resources for static data
5. Validate inputs: Ensure resource functions handle edge cases gracefully
6. Avoid exposing sensitive data: Resources are visible to all connected clients

Related Pages

Related topics: FastMCP Server Development, Context and Session Management

Tools and Structured Output

Overview

The Model Context Protocol (MCP) Python SDK provides a comprehensive system for defining and managing tools — callable functions that LLM clients can invoke to perform actions or retrieve data. Alongside tools, the SDK supports structured output through resources and well-defined result types, enabling type-safe, predictable communication between servers and clients.

Tools serve as the primary mechanism for servers to expose executable functionality to clients. Each tool has a name, optional title and description, input parameters, and a return type. The SDK enforces validation on tool names and provides mechanisms for handling errors gracefully.

Sources: src/mcp/server/mcpserver/server.py:add_tool (lines covering the add_tool method signature)

Tool Registration Architecture

Registration Flow

Tools are registered with an MCPServer instance using the add_tool() method. The registration process validates inputs, stores metadata, and prepares handlers for incoming tool calls.

graph TD
    A[Developer defines function] --> B[Call server.add_tool fn, name, title, description]
    B --> C{Validation}
    C -->|Name valid| D[Store in ToolManager]
    C -->|Invalid name| E[Raise ValueError]
    D --> F[Register with LowLevel Server]
    F --> G[Handle list_tools callback]
    
    H[Client requests tools] --> I[_handle_list_tools]
    I --> J[Return tool manifests]
    
    K[Client calls tool] --> L[_handle_call_tool]
    L --> M[Execute function]
    M --> N[Return CallToolResult]

Adding Tools to Server

Tools are added via the server's add_tool() method, which accepts the following parameters:

Sources: src/mcp/server/mcpserver/server.py:add_tool (method signature)

Tool Name Validation

The SDK enforces strict rules for tool names to ensure compatibility across clients and servers. Validation is performed according to the SEP-986 specification.

Validation Rules

Validation Function

The validate_tool_name() function returns a ToolNameValidationResult containing:

• is_valid: bool — Whether the name passes all mandatory checks
• warnings: list[str] — Non-fatal issues that may cause parsing problems

def validate_tool_name(name: str) -> ToolNameValidationResult:
    warnings: list[str] = []
    
    if not name:
        return ToolNameValidationResult(is_valid=False, warnings=["Tool name cannot be empty"])
    
    if len(name) > 128:
        return ToolNameValidationResult(
            is_valid=False,
            warnings=[f"Tool name exceeds maximum length of 128 characters"]
        )
    
    if " " in name:
        warnings.append("Tool name contains spaces, which may cause parsing issues")
    
    # ... additional validation

Sources: src/mcp/shared/tool_name_validation.py:validate_tool_name

Tool Manager

The ToolManager class handles storage, retrieval, and lifecycle management of registered tools.

Core Responsibilities

Manager Configuration

Structured Output

The SDK provides mechanisms for returning structured data from tools and resources.

CallToolResult

The standard result type for tool calls, containing:

class CallToolResult(BaseModel):
    content: list[TextContent | ImageContent | EmbeddedResource]
    is_error: bool | None = None

Content Types

Structured Output Example

@mcp.tool()
def get_user_info(user_id: str) -> CallToolResult:
    user = fetch_user(user_id)
    return CallToolResult(
        content=[
            TextContent(
                type="text",
                text=f"User: {user.name}\nEmail: {user.email}"
            )
        ]
    )

Sources: examples/snippets/servers/structured_output.py

Error Handling

Resource Errors

When reading resources fails, the server catches exceptions and converts them to ResourceError:

async def read_resource(self, uri: AnyUrl | str, context: Context | None = None):
    try:
        resource = await self._resource_manager.get_resource(uri, context)
    except ValueError as exc:
        raise ResourceError(f"Unknown resource: {uri}") from exc
    
    try:
        content = await resource.read()
        return [ReadResourceContents(content=content, mime_type=resource.mime_type)]
    except Exception as exc:
        logger.exception(f"Error getting resource {uri}")
        raise ResourceError(f"Error reading resource {uri}") from exc

Error Logging Guidelines

Sources: src/mcp/server/mcpserver/server.py:read_resource and AGENTS.md:Exception Handling

Resource Types

Resources provide structured data access alongside tools.

FunctionResource

A resource backed by a callable function:

@FunctionResource.from_function(
    uri="file://config/{env}",
    name="config_loader",
    title="Configuration Loader",
    description="Load configuration for specified environment",
    mime_type="application/json",
)
def load_config(env: str) -> dict:
    return {"environment": env, "debug": True}

FileResource

A resource that reads directly from the filesystem:

class FileResource(Resource):
    path: Path  # Must be absolute
    is_binary: bool = False
    mime_type: str = "text/plain"

Sources: src/mcp/server/mcpserver/resources/types.py:FunctionResource and src/mcp/server/mcpserver/resources/types.py:FileResource

Direct Tool Result Handling

For advanced use cases, tools can return results directly without wrapping in CallToolResult:

@mcp.tool()
def direct_result_tool(arg: str) -> str:
    """Return a string directly as tool output."""
    return f"Processed: {arg}"

The SDK automatically wraps primitive returns in appropriate content types.

Sources: examples/snippets/servers/direct_call_tool_result.py

Tool Calling Workflow

sequenceDiagram
    participant Client
    participant MCPServer
    participant ToolManager
    participant ToolFunction
    
    Client->>MCPServer: list_tools()
    MCPServer->>ToolManager: list_tools()
    ToolManager-->>MCPServer: [Tool manifests]
    MCPServer-->>Client: Tool list
    
    Client->>MCPServer: call_tool(name, arguments)
    MCPServer->>ToolManager: get_tool(name)
    ToolManager->>ToolFunction: execute(arguments)
    ToolFunction-->>ToolManager: result
    ToolManager-->>MCPServer: CallToolResult
    MCPServer-->>Client: Result

Best Practices

1. Use descriptive names: Tool names should clearly indicate their purpose
2. Add docstrings: All public tool functions should have documentation
3. Handle errors gracefully: Catch specific exceptions rather than using bare except
4. Return structured data: Prefer structured output over plain text when appropriate
5. Validate inputs: Use Pydantic models or similar for input validation
6. Log appropriately: Use logger.exception() in exception handlers

Sources: AGENTS.md:Code Quality and AGENTS.md:Exception Handling

Related Pages

Related topics: FastMCP Server Development, Resources

Prompts and Templates

Overview

Prompts and Templates in the Model Context Protocol (MCP) SDK enable servers to expose reusable prompt configurations that clients can retrieve and use with dynamic arguments. This feature allows MCP servers to define structured, parameterized prompts that can include text content, embedded resources, and complex message structures.

The prompts system provides a declarative way to define prompt templates server-side, which clients can then consume through the MCP protocol's get_prompt and list_prompts operations.

Architecture

Core Components

graph TD
    A[Client] -->|list_prompts / get_prompt| B[MCPServer]
    B --> C[PromptManager]
    C --> D[Prompt Registry]
    D --> E[Prompt Instances]
    C --> F[PromptManager.get_prompt]
    F --> G[Prompt.render]
    G --> H[GetPromptResult]
    H --> A
    
    style A fill:#e1f5fe
    style B fill:#fff3e0
    style C fill:#e8f5e9

Data Flow

sequenceDiagram
    participant Client
    participant MCPServer
    participant PromptManager
    participant Prompt
    participant Context

    Client->>MCPServer: list_prompts()
    MCPServer->>PromptManager: list_prompts()
    PromptManager-->>MCPServer: List[MCPPrompt]
    MCPServer-->>Client: Prompt metadata (name, description, arguments)

    Client->>MCPServer: get_prompt(name, arguments)
    MCPServer->>PromptManager: get_prompt(name)
    PromptManager->>Prompt: Find by name
    Prompt-->>PromptManager: Prompt instance
    PromptManager->>Prompt: render(arguments, context)
    Prompt-->>PromptManager: Rendered messages
    PromptManager-->>MCPServer: GetPromptResult
    MCPServer-->>Client: GetPromptResult with messages

Prompt Registration

Using the `@prompt()` Decorator

The primary way to register prompts is through the @prompt() decorator on the MCPServer instance.

from mcp.server.mcpserver import MCPServer, Server
from mcp.types import UserMessage, TextContent

server = MCPServer(name="my-server")

@server.prompt()
def greeting() -> list[UserMessage]:
    """A simple greeting prompt."""
    return [
        UserMessage(
            role="user",
            content=TextContent(type="text", text="Hello! How can I assist you today?")
        )
    ]

Sources: src/mcp/server/mcpserver/server.py:1-300

Prompt with Arguments

Prompts can accept arguments that are passed when the client requests the prompt:

@server.prompt()
def personalized_greeting(name: str, context: str | None = None) -> list[UserMessage]:
    """A personalized greeting with optional context."""
    base_message = f"Hello, {name}!"
    if context:
        base_message += f"\n\nContext: {context}"
    
    return [
        UserMessage(
            role="user",
            content=TextContent(type="text", text=base_message)
        )
    ]

Sources: examples/servers/simple-prompt/mcp_simple_prompt/server.py:1-100

Programmatic Registration

Alternatively, prompts can be added programmatically using the add_prompt() method:

from mcp.types import Prompt, PromptArgument

prompt = Prompt(
    name="custom-prompt",
    description="A programmatically registered prompt",
    arguments=[
        PromptArgument(
            name="topic",
            description="The topic to discuss",
            required=True
        )
    ],
    # The handler is set via on_get_prompt callback
)
server.add_prompt(prompt)

Prompt Manager

The PromptManager handles the internal storage and retrieval of prompts:

class PromptManager:
    def __init__(self, warn_on_duplicate_prompts: bool = True) -> None:
        self._prompts: dict[str, Prompt] = {}
        self._warn_on_duplicate_prompts = warn_on_duplicate_prompts

    def add_prompt(self, prompt: Prompt) -> None:
        """Add a prompt to the manager."""
        ...

    def get_prompt(self, name: str) -> Prompt | None:
        """Get a prompt by name."""
        ...

    def list_prompts(self) -> list[Prompt]:
        """List all registered prompts."""
        ...

Sources: src/mcp/server/fastmcp/prompts/manager.py:1-100

Manager Configuration

Prompt Data Model

MCPPrompt Structure

@dataclass
class MCPPrompt:
    name: str                          # Unique identifier for the prompt
    description: str                  # Human-readable description
    arguments: list[MCPPromptArgument] # Parameter definitions
    icons: list[Icon] | None          # Optional icons
    annotations: Annotations | None    # Optional annotations

MCPPromptArgument Structure

@dataclass
class MCPPromptArgument:
    name: str         # Parameter name
    description: str # Human-readable description
    required: bool   # Whether the argument is mandatory

Sources: src/mcp/server/mcpserver/server.py:1-100

Working with Resources in Prompts

Prompts can embed resources directly into the returned messages:

from mcp.types import (
    UserMessage,
    TextContent,
    EmbeddedResource,
    TextResourceContents,
)

@server.prompt()
def prompt_with_resource(resource_uri: str) -> list[UserMessage]:
    """A prompt that includes an embedded resource."""
    return [
        UserMessage(
            role="user",
            content=EmbeddedResource(
                type="resource",
                resource=TextResourceContents(
                    uri=resource_uri,
                    mime_type="text/plain",
                    text="Embedded resource content for testing.",
                ),
            ),
        ),
        UserMessage(
            role="user",
            content=TextContent(
                type="text",
                text="Please process the embedded resource above."
            )
        ),
    ]

Sources: examples/servers/everything-server/mcp_everything_server/server.py:1-200

Image Content in Prompts

Prompts can include image content using ImageContent:

from mcp.types import UserMessage, ImageContent

TEST_IMAGE_BASE64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="

@server.prompt()
def prompt_with_image() -> list[UserMessage]:
    """A prompt that includes image content."""
    return [
        UserMessage(
            role="user",
            content=ImageContent(
                type="image",
                data=TEST_IMAGE_BASE64,
                mime_type="image/png"
            )
        ),
        UserMessage(
            role="user",
            content=TextContent(type="text", text="Please analyze the image above.")
        ),
    ]

Sources: examples/servers/everything-server/mcp_everything_server/server.py:1-200

Server-Side Prompt Handlers

Manual Handler Implementation

For more control, you can implement prompt handlers manually using the low-level Server class:

import types
from mcp.server import Server, ServerRequestContext

async def handle_list_prompts(
    ctx: ServerRequestContext, 
    params: types.ListPromptsRequestParams
) -> types.ListPromptsResult:
    """List all available prompts."""
    return types.ListPromptsResult(
        prompts=[
            types.Prompt(
                name="simple",
                description="A simple prompt with optional arguments",
                arguments=[
                    types.PromptArgument(
                        name="context",
                        description="Additional context to consider",
                        required=False,
                    ),
                    types.PromptArgument(
                        name="topic",
                        description="Specific topic to focus on",
                        required=False,
                    ),
                ],
            )
        ]
    )

async def handle_get_prompt(
    ctx: ServerRequestContext, 
    params: types.GetPromptRequestParams
) -> types.GetPromptResult:
    """Get a specific prompt with arguments."""
    if params.name != "simple":
        raise ValueError(f"Unknown prompt: {params.name}")

    arguments = params.arguments or {}

    return types.GetPromptResult(
        messages=create_messages(
            context=arguments.get("context"),
            topic=arguments.get("topic")
        ),
        description="A simple prompt with optional context and topic arguments",
    )

# Create server with handlers
app = Server(
    "mcp-simple-prompt",
    on_list_prompts=handle_list_prompts,
    on_get_prompt=handle_get_prompt,
)

Sources: examples/servers/simple-prompt/mcp_simple_prompt/server.py:1-100

API Reference

MCPServer Methods

GetPromptResult

@dataclass
class GetPromptResult:
    description: str                    # Prompt description
    messages: list[BaseMessage | dict]  # Rendered message content

Context Integration

Prompts can access server context for dynamic rendering:

async def get_prompt(
    self, 
    name: str, 
    arguments: dict[str, Any] | None = None, 
    context: Context[LifespanResultT, Any] | None = None
) -> GetPromptResult:
    """Get a prompt by name with arguments."""
    if context is None:
        context = Context(mcp_server=self)
    
    try:
        prompt = self._prompt_manager.get_prompt(name)
        if not prompt:
            raise ValueError(f"Unknown prompt: {name}")

        messages = await prompt.render(arguments, context)

        return GetPromptResult(
            description=prompt.description,
            messages=pydantic_core.to_jsonable_python(messages),
        )
    except Exception as e:
        logger.exception(f"Error getting prompt {name}")
        raise ValueError(str(e)) from e

Sources: src/mcp/server/mcpserver/server.py:1-100

Error Handling

The prompt system handles errors gracefully:

graph TD
    A[get_prompt request] --> B{Prompt found?}
    B -->|No| C[ValueError: Unknown prompt]
    B -->|Yes| D[Render prompt]
    D --> E{Render successful?}
    E -->|No| F[Log exception]
    E -->|Yes| G[Return GetPromptResult]
    F --> H[ResourceError to client]
    
    style C fill:#ffcdd2
    style H fill:#ffcdd2
    style G fill:#c8e6c9

Errors during prompt retrieval are logged and converted to user-friendly ValueError or ResourceError exceptions.

Best Practices

1. Provide clear argument descriptions: Document each parameter's purpose in the PromptArgument.description field.

1. Mark required arguments appropriately: Set required=True for mandatory parameters to help clients validate input.

1. Use consistent naming: Follow snake_case for prompt and argument names for consistency with Python conventions.

1. Handle missing arguments gracefully: When arguments are optional, provide sensible defaults in your render logic.

1. Include embedded resources carefully: Only include resources that are relevant to the prompt's purpose to minimize payload size.

Summary

The Prompts and Templates system in MCP provides a flexible mechanism for servers to expose reusable, parameterized prompt configurations. Through the @prompt() decorator, programmatic add_prompt() method, or manual handler implementation, developers can define prompts that support:

• Static text content
• Dynamic arguments
• Embedded resources
• Image and multimedia content
• Context-aware rendering

This abstraction enables clean separation between prompt definition and execution, making it easy to maintain and version prompt templates within the MCP server codebase.

Related Pages

Related topics: FastMCP Server Development, Server Lifecycle and Context Management

Context and Session Management

Overview

Context and Session Management form the foundational communication layer in the MCP Python SDK. The SDK provides a layered architecture where sessions encapsulate client-server communication and contexts propagate request-scoped information through the handler chain. Sessions handle the underlying transport and protocol mechanics, while contexts provide a convenient interface for servers to access request metadata, respond to clients, and coordinate complex workflows like task elicitation and sampling.

Architecture Overview

graph TB
    subgraph "Client Layer"
        Client[MCP Client]
    end
    
    subgraph "Server Layer"
        MCPServer[MCPServer]
        FastMCP[FastMCP]
    end
    
    subgraph "Session Layer"
        ServerSession[ServerSession<br/>src/mcp/server/session.py]
        SharedSession[SharedSession<br/>src/mcp/shared/session.py]
    end
    
    subgraph "Context Layer"
        Context[Context<br/>Request Context]
        TaskContext[TaskContext<br/>Task Execution Context]
    end
    
    Client <-->|Transport| ServerSession
    MCPServer --> ServerSession
    FastMCP --> ServerSession
    ServerSession --> SharedSession
    SharedSession --> Context
    Context --> TaskContext

Session Management

Session Types

The SDK implements two primary session classes that work together to provide the complete MCP communication interface.

ServerSession

The ServerSession class is the primary interface servers use to interact with clients. It manages the protocol lifecycle, handles request/response routing, and provides methods for sending notifications and responses.

Key Responsibilities:

• Route incoming requests to appropriate handlers
• Send responses and notifications back to clients
• Manage request/response correlations via request IDs
• Handle protocol-level operations like initialization

Core Methods:

class ServerSession:
    """Server-side session for MCP protocol communication."""
    
    async def send_response(self, request_id: RequestId, response: Any) -> None:
        """Send a response to a specific request."""
        
    async def send_notification(self, method: str, params: Any) -> None:
        """Send a notification without expecting a response."""
        
    async def request(self, method: str, params: Any) -> Any:
        """Send a request to the client and wait for response."""

SharedSession

The SharedSession provides the underlying message queue and state management that both client and server sessions rely upon. It implements the request/resolver pattern for correlating messages.

sequenceDiagram
    participant Server as ServerSession
    participant Shared as SharedSession
    participant Queue as MessageQueue
    participant Resolver as Resolver
    
    Server->>Shared: _build_request(messages, params)
    Shared->>Queue: enqueue(request)
    Shared->>Resolver: Create resolver for request_id
    Queue-->>Shared: Acknowledged
    Shared-->>Server: Queued message
    
    Note over Server,Resolver: Waiting for response...
    
    Resolver->>Server: response_data

Session Initialization

Sessions are initialized during the server's lifespan and injected into the request context chain:

# src/mcp/server/mcpserver/server.py
async def read_resource(
    self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
) -> Iterable[ReadResourceContents]:
    """Read a resource by URI."""
    if context is None:
        context = Context(mcp_server=self)
    # ...

Context Management

Context Class

The Context class provides request-scoped access to server functionality. It wraps a session and server reference, exposing typed methods for common operations.

class Context(Generic[LifespanResultT, SessionT]):
    """Request context providing access to server and session functionality."""
    
    def __init__(
        self,
        mcp_server: McpServer[LifespanResultT] | None = None,
        session: SessionT | None = None,
        request_id: RequestId | None = None,
        session_id: str | None = None,
    ):
        self._mcp_server = mcp_server
        self._session = session
        self._request_id = request_id
        self._session_id = session_id

Context Request Flow

graph LR
    A[Incoming Request] --> B[ServerSession]
    B --> C[Create Context]
    C --> D[Handler with Context]
    D --> E[Business Logic]
    E --> F[ctx.session.send_response]
    F --> G[Client Response]

Context Methods

The Context class exposes several key methods for server handlers:

Example Usage in Handlers:

# src/mcp/server/mcpserver/server.py
async def read_resource(
    self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
) -> Iterable[ReadResourceContents]:
    """Read a resource by URI."""
    if context is None:
        context = Context(mcp_server=self)
    try:
        resource = await self._resource_manager.get_resource(uri, context)
    except ValueError as exc:
        raise ResourceError(f"Unknown resource: {uri}") from exc

Task Context

The TaskContext class (in src/mcp/server/experimental/task_context.py) extends the base context for task-based operations, enabling complex workflows like elicitation and sampling.

graph TD
    TC[TaskContext] --> C[Context]
    TC --> QM[Queue Management]
    TC --> EP[Elicitation Provider]
    TC --> SP[Sampling Provider]
    
    EP --> ER[ElicitResult]
    SP --> CM[CreateMessageResult]

TaskContext Creation

Tasks are created with an associated context that manages the lifecycle:

# src/mcp/server/experimental/task_context.py
async def create_task_with_context(
    self,
    messages: list[types.BaseMessage],
    task_metadata: TaskMetadata | None = None,
    ttl: int | None = None,
) -> tuple[TaskContext, CreateTaskResult]:

Task Request Building

Task contexts build requests with special task metadata for task-augmented sampling:

# src/mcp/server/experimental/task_context.py
# Build request WITH task field for task-augmented sampling
request = self._session._build_create_message_request(
    messages=messages,
    max_tokens=max_tokens,
    system_prompt=system_prompt,
    include_context=include_context,
    temperature=temperature,
    stop_sequences=stop_sequences,
    metadata=metadata,
    model_preferences=model_preferences,
    tools=tools,
    tool_choice=tool_choice,
    related_task_id=self.task_id,
    task=TaskMetadata(ttl=ttl),
)

Elicitation Pattern

Elicitation allows servers to request input from users through the client. This pattern is essential for interactive workflows requiring human confirmation or input.

Elicitation Flow

sequenceDiagram
    participant Server
    participant Session
    participant Client
    participant User
    
    Server->>Session: task.elicit(params)
    Session->>Client: elicitation request
    Client->>User: Prompt for input
    User-->>Client: User response
    Client->>Session: ElicitResult
    Session-->>Server: Elicitation response

Elicitation Implementation

The elicitation callback pattern enables servers to request user input:

# examples/snippets/servers/elicitation.py
async def elicitation_callback(context, params) -> ElicitResult:
    """Handle elicitation requests from the server."""
    # params contains the elicitation request details
    # Return user decision
    return ElicitResult(action="accept", content={"confirm": True})

Elicitation in Interactive Tasks

The interactive task example demonstrates the complete elicitation flow:

# examples/servers/simple-task-interactive/server.py
async def handle_confirm_delete(ctx: ServerRequestContext, params):
    # Request user confirmation via elicitation
    result = await ctx.request(
        "SamplingMessage",
        {
            "method": "elicitation",
            "params": {
                "message": {
                    "role": "user",
                    "content": {
                        "type": "text",
                        "text": f"Confirm delete of '{params.arguments['file_name']}'?"
                    }
                }
            }
        }
    )

Sampling Pattern

Sampling enables servers to request LLM completions from the client. This is useful for servers that need AI capabilities but delegate the actual model execution to the client.

Sampling Flow

sequenceDiagram
    participant Server
    participant Session
    participant Client
    participant LLM
    
    Server->>Session: task.create_message(messages)
    Session->>Client: sampling request
    Client->>LLM: Request completion
    LLM-->>Client: Completion
    Client->>Session: CreateMessageResult
    Session-->>Server: Sampling response

Sampling Implementation

# examples/snippets/servers/tool_progress.py
async def sampling_callback(context, params) -> CreateMessageResult:
    """Handle sampling requests from the server."""
    return CreateMessageResult(
        model="gpt-4",
        role="assistant",
        content=[types.TextContent(type="text", text="Generated response")]
    )

Tool Progress Reporting

Sessions support progress reporting for long-running operations, allowing servers to send status updates during task execution.

# examples/snippets/servers/tool_progress.py
async def handle_long_running_task(ctx: ServerRequestContext, params):
    """Handle a long-running task with progress updates."""
    
    # Initialize task
    task = await ctx.session.experimental.create_task(
        messages=[],
        task_metadata=TaskMetadata(name="long_running_task")
    )
    
    # Send progress updates
    for i in range(5):
        await task.send_progress(
            total=5,
            current=i + 1,
            message=f"Processing step {i + 1}..."
        )

Progress Notification Flow

graph TD
    A[Start Task] --> B[Execute Step 1]
    B --> C[Send Progress]
    C --> D[Execute Step 2]
    D --> E[Send Progress]
    E --> F[Complete Task]

Session Request Correlation

The SDK uses a resolver pattern to correlate requests and responses across the transport layer.

graph LR
    A[Request ID] --> B[Resolver Map]
    B --> C[Pending Request]
    C --> D[Response Received]
    D --> E[Resolve Value]
    E --> F[Awaiter Notified]

Request Lifecycle

# src/mcp/shared/session.py
class Resolver(Generic[T]):
    """Resolver for correlating request/response pairs."""
    
    def __init__(self):
        self._value: T | None = None
        self._event = anyio.create_event()
    
    def resolve(self, value: T) -> None:
        """Resolve the request with a value."""
        self._value = value
        self._event.set()
    
    async def wait(self) -> T:
        """Wait for and return the resolved value."""
        await self._event.wait()
        return self._value

Error Handling

Context and session operations should handle errors gracefully using the logging patterns specified in the SDK guidelines.

# src/mcp/server/mcpserver/server.py
try:
    content = await resource.read()
    return [ReadResourceContents(content=content, mime_type=resource.mime_type, meta=resource.meta)]
except Exception as exc:
    logger.exception(f"Error getting resource {uri}")
    # If an exception happens when reading the resource, we should not leak the exception to the client.
    raise ResourceError(f"Error reading resource {uri}") from exc

Best Practices

Context Usage

1. Always use logger.exception() instead of logger.error() when catching exceptions

• Never include the exception in the message manually

``python except (OSError, PermissionError): # Handle file operation errors except json.JSONDecodeError: # Handle JSON parsing errors ``

1. Catch specific exceptions where possible

1. Never use bare except Exception: - unless in top-level handlers

Session Management

1. Sessions are created during the server lifespan and should not be instantiated directly
2. Use the context's session reference rather than storing session references
3. For task operations, use session.experimental methods

Task Context

1. Always provide TTL for long-running tasks
2. Use polling with poll_task() for task status monitoring
3. Handle cancellation via anyio.get_cancelled_exc_class()

Related Components

Summary

The MCP SDK's Context and Session Management system provides a robust foundation for server-client communication. Sessions handle the transport and protocol mechanics, while contexts provide a developer-friendly interface for request-scoped operations. The task and elicitation patterns built on this foundation enable sophisticated interactive workflows where servers can request user input and LLM completions through a well-defined callback mechanism.

Related Pages

Related topics: Low-Level Server Development

Transports Overview

Introduction

The Model Context Protocol (MCP) SDK supports multiple transport mechanisms for communication between clients and servers. Transports define how JSON-RPC messages are transmitted between the MCP client and server, providing abstraction over different communication paradigms.

The SDK provides two primary transport types:

Sources: examples/servers/simple-pagination/mcp_simple_pagination/server.py:49-57

Architecture

Transport Layer Position

MCP follows a layered architecture where transports sit at the communication layer:

graph TD
    A[MCP Client] --> B[Transport Layer]
    C[MCP Server] --> D[Transport Layer]
    B <-->|stdio / HTTP| D
    
    E[Application Layer] --> B
    D --> F[Business Logic]

Message Flow

All transports implement the same message protocol, enabling interoperability:

sequenceDiagram
    participant C as Client
    participant T as Transport
    participant S as Server
    
    C->>T: Initialize
    T->>S: JSON-RPC Request
    S->>T: JSON-RPC Response
    T->>C: Result

Stdio Transport

Overview

The stdio transport uses standard input and standard output streams for communication. This is ideal for local processes, command-line tools, and desktop application integrations.

Sources: examples/snippets/clients/stdio_client.py

Server Implementation

from mcp.server.stdio import stdio_server

async def arun():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

anyio.run(arun)

Sources: examples/servers/simple-pagination/mcp_simple_pagination/server.py:61-65

Client Implementation

from mcp.client.stdio import StdioServerParameters, stdio_client

async def run():
    params = StdioServerParameters(
        command="python",
        args=["server.py"],
        env={"key": "value"}
    )
    async with stdio_client(params) as client:
        # Use client...

Sources: examples/snippets/clients/stdio_client.py

StdioServerParameters

Streamable HTTP Transport

Overview

The streamable-http transport uses HTTP POST requests for sending messages and Server-Sent Events (SSE) or chunked transfer for receiving responses. This supports remote server deployments and web integrations.

Sources: examples/servers/simple-pagination/mcp_simple_pagination/server.py:54-56

Server Implementation

import uvicorn

if transport == "streamable-http":
    uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)

Sources: examples/servers/simple-pagination/mcp_simple_pagination/server.py:54-56

Server Options

The streamable_http_app() method supports the following configuration:

Sources: src/mcp/server/mcpserver/server.py

HTTP Endpoint Configuration

app = Server("mcp-simple-pagination", ...)
http_app = app.streamable_http_app(
    app_path="/custom-path",
    cors_domains=["https://example.com"],
    heartbeat=30.0
)
uvicorn.run(http_app, host="127.0.0.1", port=8000)

Client Configuration

Environment Variables

The SDK supports configuration via environment variables for HTTP transport:

Sources: examples/clients/simple-auth-client/README.md

Transport Selection

# stdio transport
transport = "stdio"
async with stdio_server() as streams:
    await app.run(streams[0], streams[1], app.create_initialization_options())

# HTTP transport
transport = "streamable-http"
uvicorn.run(app.streamable_http_app(), host="127.0.0.1", port=port)

Sources: examples/servers/simple-prompt/mcp_simple_prompt/server.py:52-65

Stream Protocol

Message Framing

Both transports use the MCP stream protocol for message encoding:

graph LR
    A[Application Data] --> B[JSON-RPC 2.0]
    B --> C[Content-Length Header]
    C --> D[Raw Bytes]

Protocol Requirements

• Messages are JSON-RPC 2.0 compliant
• HTTP transport uses Content-Length header for framing
• Stdio transport uses line-delimited JSON with content length prefix

Transport Selection Guide

CLI Integration

The MCP CLI tool supports installing servers with transport configuration:

mcp install server.py --name "my-server"

Sources: src/mcp/cli/cli.py

Environment variables specified during installation are preserved and can be used for transport configuration.

Best Practices

1. Security: Use stdio for local, trusted connections; use HTTPS for streamable-http in production
2. Error Handling: Always wrap transport initialization in try-except blocks
3. Timeouts: Configure appropriate timeouts for HTTP transport
4. Resource Cleanup: Use async context managers to ensure proper resource disposal
5. CORS: Restrict cors_domains when deploying HTTP servers

Example: Dual Transport Server

import click
import uvicorn
from mcp.server.stdio import stdio_server
from mcp.server.mcpserver import MCPServer

app = MCPServer(name="dual-transport-server", ...)

@click.command()
@click.option("--transport", type=click.Choice(["stdio", "streamable-http"]), default="stdio")
@click.option("--port", default=8000)
def main(transport: str, port: int):
    if transport == "streamable-http":
        uvicorn.run(app.streamable_http_app(), host="0.0.0.0", port=port)
    else:
        async def run():
            async with stdio_server() as streams:
                await app.run(streams[0], streams[1], app.create_initialization_options())
        anyio.run(run)

This pattern allows a single server implementation to serve both transport types based on deployment requirements.

Doramagic Pitfall Log

Doramagic extracted 16 source-linked risk signals. Review them before installing or handing real data to the project.

1. Configuration risk: Duplicate `initialize` with changed parameters can overwrite `ServerSession.client_params`

• Severity: high
• Finding: Configuration risk is backed by a source signal: Duplicate initialize with changed parameters can overwrite ServerSession.client_params. Treat it as a review item until the current version is checked.
• User impact: Users may get misleading failures or incomplete behavior unless configuration is checked carefully.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/2605

2. Configuration risk: Streamable HTTP server silently drops in-flight request when client reuses a JSON-RPC id

• Severity: high
• Finding: Configuration risk is backed by a source signal: Streamable HTTP server silently drops in-flight request when client reuses a JSON-RPC id. Treat it as a review item until the current version is checked.
• User impact: Users may get misleading failures or incomplete behavior unless configuration is checked carefully.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/2655

3. Configuration risk: streamable_http_client: one concurrent request HTTPStatusError tears down sibling requests

• Severity: high
• Finding: Configuration risk is backed by a source signal: streamable_http_client: one concurrent request HTTPStatusError tears down sibling requests. Treat it as a review item until the current version is checked.
• User impact: Users may get misleading failures or incomplete behavior unless configuration is checked carefully.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/2604

4. Maintenance risk: FastMCP crashes when tool return type uses Python 3.10+ `A | B | C` union syntax

• Severity: high
• Finding: Maintenance risk is backed by a source signal: FastMCP crashes when tool return type uses Python 3.10+ A | B | C union syntax. Treat it as a review item until the current version is checked.
• User impact: Users cannot judge support quality until recent activity, releases, and issue response are checked.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/2591

5. Security or permission risk: Feature Proposal: Secure Tool/Resource/Prompt Decorators with Auth + Encrypted I/O

• Severity: high
• Finding: Security or permission risk is backed by a source signal: Feature Proposal: Secure Tool/Resource/Prompt Decorators with Auth + Encrypted I/O. Treat it as a review item until the current version is checked.
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/1305

6. Security or permission risk: Progress notifications cause server to hang on stdio transport

• Severity: high
• Finding: Security or permission risk is backed by a source signal: Progress notifications cause server to hang on stdio transport. Treat it as a review item until the current version is checked.
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/1141

7. Security or permission risk: Streamable HTTP server accepts mismatched `MCP-Protocol-Version` header and body `protocolVersion` on `initialize`

• Severity: high
• Finding: Security or permission risk is backed by a source signal: Streamable HTTP server accepts mismatched MCP-Protocol-Version header and body protocolVersion on initialize. Treat it as a review item until the current version is checked.
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/2618

8. Security or permission risk: User-Agent header in sHTTP transport is not forwarded to auth flow

• Severity: high
• Finding: Security or permission risk is backed by a source signal: User-Agent header in sHTTP transport is not forwarded to auth flow. Treat it as a review item until the current version is checked.
• User impact: The project may affect permissions, credentials, data exposure, or host boundaries.
• Recommended check: Open the linked source, confirm whether it still applies to the current version, and keep the first run isolated.
• Evidence: Source-linked evidence: https://github.com/modelcontextprotocol/python-sdk/issues/1664

9. Installation risk: Developers should check this installation risk before relying on the project: Bug: `anyio.Lock` in `oauth2.py` raises "current task is not holding this lock" under cross-task generator driving

• Severity: medium
• Finding: Developers should check this installation risk before relying on the project: Bug: anyio.Lock in oauth2.py raises "current task is not holding this lock" under cross-task generator driving
• User impact: Developers may fail before the first successful local run: Bug: anyio.Lock in oauth2.py raises "current task is not holding this lock" under cross-task generator driving
• Recommended check: Before packaging this project, run the relevant install/config/quickstart check for: Bug: anyio.Lock in oauth2.py raises "current task is not holding this lock" under cross-task generator driving. Context: Observed when using windows
• Evidence: failure_mode_cluster:github_issue | fmev_7ab763a43233ec1e7dcaebc3255017a6 | https://github.com/modelcontextprotocol/python-sdk/issues/2644 | Bug: anyio.Lock in oauth2.py raises "current task is not holding this lock" under cross-task generator driving

10. Installation risk: Developers should check this installation risk before relying on the project: FastMCP.__init__ clobbers root logger with INFO RichHandler — hangs stdio servers under back-pressure

• Severity: medium
• Finding: Developers should check this installation risk before relying on the project: FastMCP.__init__ clobbers root logger with INFO RichHandler — hangs stdio servers under back-pressure
• User impact: Developers may fail before the first successful local run: FastMCP.__init__ clobbers root logger with INFO RichHandler — hangs stdio servers under back-pressure
• Recommended check: Before packaging this project, run the relevant install/config/quickstart check for: FastMCP.__init__ clobbers root logger with INFO RichHandler — hangs stdio servers under back-pressure. Context: Observed when using node, python, macos
• Evidence: failure_mode_cluster:github_issue | fmev_4529ab733a6558a19e9c3f318ce112c5 | https://github.com/modelcontextprotocol/python-sdk/issues/2527 | FastMCP.__init__ clobbers root logger with INFO RichHandler — hangs stdio servers under back-pressure

11. Installation risk: Developers should check this installation risk before relying on the project: Types-only install option

• Severity: medium
• Finding: Developers should check this installation risk before relying on the project: Types-only install option
• User impact: Developers may fail before the first successful local run: Types-only install option
• Recommended check: Before packaging this project, run the relevant install/config/quickstart check for: Types-only install option. Context: Observed during installation or first-run setup.
• Evidence: failure_mode_cluster:github_issue | fmev_4b31523e85ae0860236ec672488b4035 | https://github.com/modelcontextprotocol/python-sdk/issues/2581 | Types-only install option

12. Configuration risk: Developers should check this configuration risk before relying on the project: Add `invalid_target` to `AuthorizationErrorCode` (RFC 8707)

• Severity: medium
• Finding: Developers should check this configuration risk before relying on the project: Add invalid_target to AuthorizationErrorCode (RFC 8707)
• User impact: Developers may misconfigure credentials, environment, or host setup: Add invalid_target to AuthorizationErrorCode (RFC 8707)
• Recommended check: Before packaging this project, run the relevant install/config/quickstart check for: Add invalid_target to AuthorizationErrorCode (RFC 8707). Context: Observed when using python
• Evidence: failure_mode_cluster:github_issue | fmev_4c29698ade9229f6578bf641e5ca4aec | https://github.com/modelcontextprotocol/python-sdk/issues/2641 | Add invalid_target to AuthorizationErrorCode (RFC 8707)

Community Discussion Evidence

Doramagic exposes project-level community discussion separately from official documentation. Review these links before using python-sdk with real data or production workflows.

• User-Agent header in sHTTP transport is not forwarded to auth flow - github / github_issue
• Server OAuth metadata hardcodes token_endpoint_auth_methods_supported, b - github / github_issue
• Streamable HTTP server silently drops in-flight request when client reus - github / github_issue
• Progress notifications cause server to hang on stdio transport - github / github_issue
• mcp.server.fastmcp.FastMCP HTTP transport deadlocks after ~5 sequential - github / github_issue
• streamable HTTP client parses zstd-compressed JSON response bytes as JSO - github / github_issue
• MCP client does not retry authenticated request after successful OAuth t - github / github_issue
• Agents talking to MCP Server, SSL verification is failing. Has some been - github / github_issue
• Trying to connect to a MCP url got the follwing error any idea why ? - github / github_issue
• how to return images from an mcp server - github / github_issue
• Bug: anyio.Lock in oauth2.py raises "current task is not holding thi - github / github_issue
• Duplicate initialize with changed parameters can overwrite ServerSession.client_params - GitHub / issue