Skip to main content

Python Mcp Server

Instructions for building Model Context Protocol (MCP) servers using the Python SDK v2

Install

These instructions apply to the MCP Python SDK v2 and the MCP specification revision 2026-07-28.

Instructions

  • Use uv for project management: uv init --app --no-package mcp-server-demo and uv add "mcp[cli]>=2,<3".
  • Import the server class from mcp.server: from mcp.server import MCPServer.
  • Do not use FastMCP or mcp.server.fastmcp. SDK v2 removed this import path.
  • Import Context, Image, Audio, Resolve, Elicit, and the prompt message classes from mcp.server.mcpserver.
  • Use the @mcp.tool(), @mcp.resource(), and @mcp.prompt() decorators for registration.
  • Add type hints to all parameters. The SDK makes the schemas from the type hints and validates the arguments.
  • Use Pydantic models, TypedDicts, or dataclasses for structured output.
  • A tool returns structured output when its return type is compatible.
  • For stdio transport, use mcp.run() or mcp.run(transport="stdio").
  • For HTTP servers, use mcp.run(transport="streamable-http"), or mount mcp.streamable_http_app() in Starlette or FastAPI.
  • Give transport options (host, port, json_response, stateless_http, transport_security) to run(), not to MCPServer(...).
  • When you mount the server, give the same options to streamable_http_app(). This method has no port option.
  • Give the server name as the first MCPServer argument. Give all other constructor arguments as keyword arguments.
  • Set version on MCPServer. If you do not set it, the server reports an empty version.
  • The SDK v2 types use snake_case fields, for example read_only_hint, structured_content, and is_error.
  • Add a ctx: Context parameter to a tool, resource, or prompt to get the request context.
  • Report progress with await ctx.report_progress(progress, total, message).
  • Get user input with a Resolve(fn) parameter. Return Elicit(message, Model) from fn. This works with clients of all protocol revisions.
  • Do not use ctx.elicit(). It fails on a 2026-07-28 connection.
  • Log with the standard logging module. Protocol logging (ctx.info() and similar methods) is deprecated.
  • Do not use sampling (ctx.session.create_message()). It is deprecated. Call the LLM provider API directly.
  • Do not use roots (ctx.session.list_roots()). It is deprecated. Get paths from tool parameters or the server configuration.
  • Configure icons with Icon(src="https://...", mime_type="image/png") from mcp.types. Server, tools, resources, and prompts accept icons=[...].
  • Use the Image class to return images: return Image(data=png_bytes, format="png") or return Image(path=file_path).
  • Define resource templates with RFC 6570 URI templates: @mcp.resource("greeting://{name}").
  • Add completion support with the @mcp.completion() decorator.
  • Use a lifespan context manager for startup and shutdown of shared resources. For stdio and HTTP, the lifespan runs one time for the server. Each in-memory Client(mcp) runs it again.
  • Get the lifespan object in tools from ctx.request_context.lifespan_context. Use Context[AppContext] for a typed result.
  • Raise ToolError (from mcp.server.mcpserver.exceptions) for an error that the model must read.
  • Do not raise MCPError for a tool failure. The client gets a protocol error, not a tool result with is_error. Many hosts do not show this error to the model.
  • Send list change notifications with await ctx.notify_tools_changed(). Do not use ctx.session.send_tool_list_changed(). A 2026-07-28 connection drops it.
  • On the 2026-07-28 revision, Streamable HTTP requests have no session. stateless_http=True changes only how the server serves clients of earlier revisions.
  • json_response=True sends one JSON body for each request. In this mode, the server cannot send progress notifications during the request.
  • Test servers with uv run mcp dev server.py (MCP Inspector) or uv run mcp install server.py (Claude Desktop).
  • Write tests with the in-memory client: async with Client(mcp) as client: (from mcp import Client).
  • Mount more than one server in Starlette with different paths: Mount("/path", app=mcp.streamable_http_app()). The lifespan of the host app must enter mcp.session_manager.run() for each server.
  • For browser clients, configure CORS. Allow the Mcp-* request headers. Expose the Mcp-Session-Id response header.
  • If you do not set transport_security, the SDK checks the Host and Origin headers only when the host is 127.0.0.1, localhost, or ::1. For all other hosts, set transport_security=TransportSecuritySettings(allowed_hosts=[...], allowed_origins=[...]) (from mcp.server.transport_security).
  • For more than one worker, set request_state_security=RequestStateSecurity(keys=[...]) on MCPServer.
  • Use the low-level Server class (from mcp.server) only when MCPServer does not give sufficient control.

Best Practices

  • Give each tool function one responsibility.
  • Write clear docstrings. They become the tool descriptions.
  • Use descriptive parameter names with type hints.
  • Validate inputs with Pydantic Field descriptions.
  • Use async functions for I/O-bound operations. The SDK runs sync functions on a worker thread.
  • Log to stderr. Do not use print() in a stdio server, because stdout is the protocol channel.
  • Use environment variables for configuration.
  • Be careful when a tool gives access to the file system or the network.

Common Patterns

Basic Server Setup (stdio)

from mcp.server import MCPServer

mcp = MCPServer("My Server", version="0.1.0")

@mcp.tool()
def calculate(a: int, b: int, op: str) -> int:
    """Perform calculation"""
    if op == "add":
        return a + b
    return a - b

if __name__ == "__main__":
    mcp.run()  # stdio by default

HTTP Server

from mcp.server import MCPServer

mcp = MCPServer("My HTTP Server", version="0.1.0")

@mcp.tool()
def hello(name: str = "World") -> str:
    """Greet someone"""
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="127.0.0.1", port=8000)

Tool with Structured Output

from pydantic import BaseModel, Field

class WeatherData(BaseModel):
    temperature: float = Field(description="Temperature in Celsius")
    condition: str
    humidity: float

@mcp.tool()
def get_weather(city: str) -> WeatherData:
    """Get weather for a city"""
    return WeatherData(
        temperature=22.5,
        condition="sunny",
        humidity=65.0
    )

Dynamic Resource

@mcp.resource("users://{user_id}")
def get_user(user_id: str) -> str:
    """Get user profile data"""
    return f"User {user_id} profile data"

Tool with Context

import logging

from mcp.server.mcpserver import Context

logger = logging.getLogger(__name__)

@mcp.tool()
async def process_data(data: str, ctx: Context) -> str:
    """Process data with progress reports"""
    logger.info("Processing: %s", data)
    await ctx.report_progress(0.5, 1.0, "Halfway done")
    return f"Processed: {data}"

Tool with User Input

from typing import Annotated

from pydantic import BaseModel

from mcp.server.mcpserver import AcceptedElicitation, Elicit, ElicitationResult, Resolve

class Quantity(BaseModel):
    copies: int

async def ask_quantity() -> Elicit[Quantity]:
    """Ask the user how many copies to reserve."""
    return Elicit("How many copies?", Quantity)

@mcp.tool()
async def reserve(
    title: str,
    quantity: Annotated[ElicitationResult[Quantity], Resolve(ask_quantity)],
) -> str:
    """Reserve copies of a book"""
    if isinstance(quantity, AcceptedElicitation):
        return f"Reserved {quantity.data.copies} of {title!r}."
    return "Nothing reserved."

Lifespan Management

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from mcp.server import MCPServer
from mcp.server.mcpserver import Context

@dataclass
class AppContext:
    db: Database  # Your database client

@asynccontextmanager
async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
    db = await Database.connect()
    try:
        yield AppContext(db=db)
    finally:
        await db.disconnect()

mcp = MCPServer("My App", version="0.1.0", lifespan=app_lifespan)

@mcp.tool()
def query(sql: str, ctx: Context[AppContext]) -> str:
    """Query database"""
    db = ctx.request_context.lifespan_context.db
    return db.execute(sql)

Prompt with Messages

from mcp.server.mcpserver import AssistantMessage, Message, UserMessage

@mcp.prompt(title="Code Review")
def review_code(code: str) -> list[Message]:
    """Create code review prompt"""
    return [
        UserMessage("Review this code:"),
        UserMessage(code),
        AssistantMessage("I'll review the code for you.")
    ]

Error Handling

from mcp.server.mcpserver.exceptions import ToolError

@mcp.tool()
async def risky_operation(input: str) -> str:
    """Operation that might fail"""
    if not input:
        raise ToolError("Input must not be empty.")
    result = await perform_operation(input)  # Your operation
    return f"Success: {result}"

In-Memory Test

import pytest

from mcp import Client

from server import mcp  # The HTTP Server example

@pytest.mark.anyio
async def test_hello():
    async with Client(mcp) as client:
        result = await client.call_tool("hello", {"name": "MCP"})
        assert result.structured_content == {"result": "Hello, MCP!"}