MCP 协议设计与实现

第10章 Python Server 实现剖析

作者杨艺韬 · 8,382 字

第10章 Python Server 实现剖析

在前面的章节中，我们已经深入分析了 TypeScript SDK 的服务端实现。本章将转向 Python SDK，剖析其服务端的核心架构与实现细节。Python SDK 的服务端实现与 TypeScript SDK 在协议层面保持一致，但在工程实践上充分利用了 Python 生态的优势：Pydantic 提供类型安全的数据验证，anyio 提供跨异步框架的并发抽象，Starlette 提供生产级的 ASGI 集成。理解这些实现细节，不仅有助于我们更好地使用 Python SDK 构建 MCP 服务，也能帮助我们在遇到问题时快速定位和解决。

10.1 双层架构：MCPServer 与 Low-Level Server

Python SDK 的服务端采用了清晰的双层架构设计。底层是 Server 类（位于 mcp.server.lowlevel.server），提供基于回调函数的原始接口；上层是 MCPServer 类（位于 mcp.server.mcpserver.server），提供基于装饰器的开发体验。这种分层与 TypeScript SDK 的单层 Server 类形成鲜明对比。

graph TB
    subgraph "用户代码层"
        A["@server.tool()"] --> B["@server.resource()"]
        B --> C["@server.prompt()"]
    end

    subgraph "高层 API (MCPServer)"
        D[ToolManager] --> G[MCPServer]
        E[ResourceManager] --> G
        F[PromptManager] --> G
    end

    subgraph "底层 API (Low-Level Server)"
        H["Server (lowlevel)"]
        H --> I["request_handlers dict"]
        H --> J["notification_handlers dict"]
    end

    subgraph "传输层"
        K[stdio_server]
        L[SseServerTransport]
        M[StreamableHTTPSessionManager]
    end

    subgraph "会话层"
        N[ServerSession]
    end

    A --> D
    B --> E
    C --> F
    G -->|"委托"| H
    H --> N
    N --> K
    N --> L
    N --> M

底层 Server 类在构造时通过 on_* 参数接收处理函数：

class Server(Generic[LifespanResultT]):
    def __init__(
        self,
        name: str,
        *,
        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_prompts: Callable[...] | None = None,
        on_get_prompt: Callable[...] | None = None,
        lifespan: Callable[...] = lifespan,
        ...
    ):
        self._request_handlers: dict[str, Callable[...]] = {}
        # 将 on_* 参数映射到方法字符串
        self._request_handlers.update({
            method: handler
            for method, handler in {
                "tools/list": on_list_tools,
                "tools/call": on_call_tool,
                "resources/list": on_list_resources,
                "resources/read": on_read_resource,
                "prompts/list": on_list_prompts,
                "prompts/get": on_get_prompt,
                ...
            }.items()
            if handler is not None
        })

这段代码的关键在于，Server 类本身不关心工具、资源、提示词的具体管理逻辑，它只是一个「请求分发器」。收到 "tools/list" 请求就调用对应的 handler，收到 "tools/call" 请求就调用另一个 handler。这种设计使得底层 Server 保持极度简洁。

高层 MCPServer 类则在内部创建了底层 Server 实例，并将自己的私有方法注册为各个 handler：

class MCPServer(Generic[LifespanResultT]):
    def __init__(self, name: str | None = None, ...):
        self._tool_manager = ToolManager(...)
        self._resource_manager = ResourceManager(...)
        self._prompt_manager = PromptManager(...)
        self._lowlevel_server = Server(
            name=name or "mcp-server",
            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_prompts=self._handle_list_prompts,
            on_get_prompt=self._handle_get_prompt,
            lifespan=...,
        )

这种双层设计带来了两个好处。第一，普通开发者使用 MCPServer 的装饰器 API 即可快速开发，无需了解协议细节。第二，需要更细粒度控制的高级用户可以直接使用底层 Server 类，自行实现所有 handler 逻辑。

10.2 装饰器体系：@tool、@resource、@prompt

MCPServer 最核心的 API 就是三个装饰器。它们的设计哲学是：让函数签名即协议契约。

10.2.1 @tool 装饰器

@tool() 装饰器的实现非常精巧：

def tool(
    self,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    ...
) -> Callable[[_CallableT], _CallableT]:
    # 防止误用：@tool 而不是 @tool()
    if callable(name):
        raise TypeError(
            "The @tool decorator was used incorrectly. "
            "Did you forget to call it? Use @tool() instead of @tool"
        )

    def decorator(fn: _CallableT) -> _CallableT:
        self.add_tool(fn, name=name, title=title, description=description, ...)
        return fn

    return decorator

注意 if callable(name) 这个防御性检查。当开发者写 @server.tool（少了括号）时，Python 会把被装饰的函数作为 name 参数传入，此时 name 是一个 callable，SDK 会抛出明确的错误信息，而不是产生令人困惑的运行时行为。

装饰器的核心工作委托给了 add_tool 方法，进而委托给 ToolManager.add_tool，最终调用 Tool.from_function 完成函数到工具的转换：

@classmethod
def from_function(cls, fn, name=None, description=None, ...):
    func_name = name or fn.__name__
    func_doc = description or fn.__doc__ or ""
    is_async = is_async_callable(fn)

    # 自动检测 Context 参数
    context_kwarg = find_context_parameter(fn)

    # 利用 Pydantic 从函数签名生成 JSON Schema
    func_arg_metadata = func_metadata(
        fn,
        skip_names=[context_kwarg] if context_kwarg else [],
    )
    parameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True)

    return cls(fn=fn, name=func_name, parameters=parameters, ...)

这段代码展示了几个关键的设计决策：

名称推断：如果不显式指定 name，使用函数名（fn.__name__）。
描述推断：如果不显式指定 description，使用函数的 docstring。
Context 自动注入：通过 find_context_parameter 检测函数签名中是否有 Context 类型的参数，如果有，则在调用时自动注入，而不把它暴露给 JSON Schema。
Pydantic Schema 生成：利用 func_metadata 将函数签名转化为 Pydantic Model，再通过 model_json_schema 生成符合 MCP 协议的 input_schema。

10.2.2 @resource 装饰器

@resource() 装饰器的设计更为复杂，因为它需要区分静态资源和模板资源：

def resource(self, uri: str, *, name=None, description=None, mime_type=None, ...):
    def decorator(fn):
        sig = inspect.signature(fn)
        has_uri_params = "{" in uri and "}" in uri
        has_func_params = bool(sig.parameters)

        if has_uri_params or has_func_params:
            # URI 中有 {param} 占位符，注册为模板资源
            uri_params = set(re.findall(r"{(\w+)}", uri))
            func_params = {p for p in sig.parameters.keys() if p != context_param}

            if uri_params != func_params:
                raise ValueError(
                    f"Mismatch between URI parameters {uri_params} "
                    f"and function parameters {func_params}"
                )
            self._resource_manager.add_template(fn=fn, uri_template=uri, ...)
        else:
            # 无参数，注册为静态资源
            resource = FunctionResource.from_function(fn=fn, uri=uri, ...)
            self.add_resource(resource)
        return fn
    return decorator

这种设计使得同一个装饰器可以同时处理两种场景：

# 静态资源 —— 函数无参数
@server.resource("config://app-settings")
def get_settings() -> str:
    return json.dumps({"theme": "dark", "language": "zh"})

# 模板资源 —— URI 包含参数，函数签名与之匹配
@server.resource("users://{user_id}/profile")
async def get_user_profile(user_id: str) -> str:
    return await fetch_user(user_id)

SDK 会在注册时验证 URI 模板中的参数名与函数参数名是否一致，不一致则立即报错，而非等到运行时才发现问题。

10.2.3 @prompt 装饰器

@prompt() 装饰器的实现相对简洁，它通过 Prompt.from_function 从函数签名中提取参数信息：

def prompt(self, name=None, title=None, description=None, ...):
    def decorator(func):
        prompt = Prompt.from_function(
            func, name=name, title=title, description=description
        )
        self.add_prompt(prompt)
        return func
    return decorator

10.3 Pydantic 深度集成

Python SDK 对 Pydantic 的使用远不止于数据验证，而是将其作为整个类型系统的基石。

10.3.1 函数签名到 JSON Schema 的自动转换

func_metadata 函数是 Pydantic 集成的核心。它接受一个普通 Python 函数，解析其类型注解，动态构建一个 Pydantic Model，再通过该 Model 生成 JSON Schema。

graph LR
    A["Python 函数签名<br/>def add(a: int, b: float)"] --> B["inspect.signature()"]
    B --> C["提取参数类型注解"]
    C --> D["动态创建 Pydantic Model<br/>class AddArgs(BaseModel):<br/>    a: int<br/>    b: float"]
    D --> E["model_json_schema()"]
    E --> F["JSON Schema<br/>{type: 'object',<br/> properties: {<br/>   a: {type: 'integer'},<br/>   b: {type: 'number'}<br/> }}"]

这意味着开发者只需要写标准的 Python 类型注解，SDK 就能自动生成符合 MCP 协议要求的 JSON Schema，无需手动编写任何 Schema 定义。

@server.tool()
async def query_database(
    table: str,
    limit: int = 10,
    filters: dict[str, str] | None = None,
) -> str:
    """查询数据库中的数据"""
    ...

上面的代码会自动生成如下 input_schema：

{
  "type": "object",
  "properties": {
    "table": {"type": "string"},
    "limit": {"type": "integer", "default": 10},
    "filters": {
      "anyOf": [
        {"type": "object", "additionalProperties": {"type": "string"}},
        {"type": "null"}
      ],
      "default": null
    }
  },
  "required": ["table"]
}

10.3.2 参数验证与调用

当工具被调用时，FuncMetadata.call_fn_with_arg_validation 会使用 Pydantic Model 对传入参数进行验证，然后再调用实际函数：

# Tool.run 方法
async def run(self, arguments, context, convert_result=False):
    try:
        result = await self.fn_metadata.call_fn_with_arg_validation(
            self.fn,
            self.is_async,
            arguments,
            {self.context_kwarg: context} if self.context_kwarg else None,
        )
        if convert_result:
            result = self.fn_metadata.convert_result(result)
        return result
    except UrlElicitationRequiredError:
        raise
    except Exception as e:
        raise ToolError(f"Error executing tool {self.name}: {e}") from e

Pydantic 验证在这里提供了双重保障：类型不匹配时会给出清晰的错误信息，而不是在函数内部产生难以追踪的运行时错误。

10.3.3 Context 类的 Pydantic 基类设计

一个值得注意的设计选择是，Context 类继承自 pydantic.BaseModel：

class Context(BaseModel, Generic[LifespanContextT, RequestT]):
    _request_context: ServerRequestContext | None
    _mcp_server: MCPServer | None

使用下划线前缀的字段在 Pydantic v2 中是私有字段（不参与序列化），这是一种有意为之的设计：Context 对象需要在内部传递复杂的运行时状态（session、server 引用等），但这些状态不应该被意外序列化或暴露给外部。

10.4 anyio 异步模型

Python SDK 选择 anyio 而非直接使用 asyncio，这是一个影响深远的架构决策。

10.4.1 为什么是 anyio

anyio 是一个异步兼容层，能同时运行在 asyncio 和 trio 之上。选择 anyio 意味着：

支持 asyncio 后端：这是 Python 异步编程的主流选择。
支持 trio 后端：trio 提供了更严格的结构化并发模型。
结构化并发原语：anyio.create_task_group() 强制所有子任务在退出作用域前完成或取消。

10.4.2 结构化并发在消息处理中的应用

底层 Server.run 方法展示了 anyio 结构化并发的核心用法：

async def run(self, read_stream, write_stream, initialization_options, ...):
    async with AsyncExitStack() as stack:
        lifespan_context = await stack.enter_async_context(self.lifespan(self))
        session = await stack.enter_async_context(
            ServerSession(read_stream, write_stream, initialization_options, ...)
        )

        async with anyio.create_task_group() as tg:
            try:
                async for message in session.incoming_messages:
                    context = contextvars.copy_context()
                    context.run(
                        tg.start_soon,
                        self._handle_message,
                        message, session, lifespan_context, raise_exceptions,
                    )
            finally:
                tg.cancel_scope.cancel()

这段代码有几个关键的设计要点：

AsyncExitStack 管理资源生命周期，确保 lifespan 和 session 的正确清理。
anyio.create_task_group 为每个入站消息创建并发处理任务。
contextvars.copy_context 确保每个消息处理任务继承正确的上下文（如 OpenTelemetry trace context）。
finally 块中的 cancel 在传输关闭时取消所有正在进行的 handler，防止它们尝试向已关闭的 write stream 发送响应。

sequenceDiagram
    participant Transport as 传输层
    participant Session as ServerSession
    participant Server as Server.run()
    participant TG as TaskGroup
    participant H1 as Handler 1
    participant H2 as Handler 2

    Server->>Session: enter_async_context(session)
    Session->>Transport: 建立连接

    loop 消息循环
        Transport->>Session: incoming_message
        Session->>Server: yield message
        Server->>TG: tg.start_soon(handle_message)
        TG->>H1: 并发处理请求 A
        Transport->>Session: incoming_message
        Session->>Server: yield message
        Server->>TG: tg.start_soon(handle_message)
        TG->>H2: 并发处理请求 B
        H1->>Session: respond(result_A)
        H2->>Session: respond(result_B)
    end

    Transport->>Session: 连接关闭
    Server->>TG: cancel_scope.cancel()
    TG->>H1: 取消进行中的任务
    TG->>H2: 取消进行中的任务

10.4.3 stdio 传输的 anyio 实现

stdio 传输是最简单的传输实现，展示了 anyio 流式处理的典型模式：

@asynccontextmanager
async def stdio_server(stdin=None, stdout=None):
    read_stream_writer, read_stream = create_context_streams(0)
    write_stream, write_stream_reader = create_context_streams(0)

    async def stdin_reader():
        async with read_stream_writer:
            async for line in stdin:
                message = types.jsonrpc_message_adapter.validate_json(line)
                await read_stream_writer.send(SessionMessage(message))

    async def stdout_writer():
        async with write_stream_reader:
            async for session_message in write_stream_reader:
                json = session_message.message.model_dump_json(...)
                await stdout.write(json + "\n")
                await stdout.flush()

    async with anyio.create_task_group() as tg:
        tg.start_soon(stdin_reader)
        tg.start_soon(stdout_writer)
        yield read_stream, write_stream

两个关键点：第一，使用 create_context_streams(0) 创建无缓冲的内存流，确保背压传导。第二，stdin_reader 和 stdout_writer 作为两个并发任务运行在同一个 TaskGroup 中，任何一个出错都会取消另一个。

10.5 Starlette/ASGI 集成

Python SDK 的 HTTP 传输建立在 Starlette 之上，这使得 MCP 服务可以作为标准的 ASGI 应用部署。

10.5.1 多传输统一入口

MCPServer.run() 是同步入口方法，它根据传输类型分派到不同的异步实现：

def run(self, transport="stdio", **kwargs):
    match transport:
        case "stdio":
            anyio.run(self.run_stdio_async)
        case "sse":
            anyio.run(lambda: self.run_sse_async(**kwargs))
        case "streamable-http":
            anyio.run(lambda: self.run_streamable_http_async(**kwargs))

注意 anyio.run 的使用。它会创建一个新的事件循环并运行整个服务。对于 SSE 和 Streamable HTTP 传输，最终都是通过构建 Starlette 应用并用 uvicorn 启动来实现的。

10.5.2 Starlette 应用构建

以 Streamable HTTP 为例，streamable_http_app() 方法构建了完整的 ASGI 应用：

def streamable_http_app(self, *, streamable_http_path="/mcp", ...):
    session_manager = StreamableHTTPSessionManager(
        app=self, event_store=event_store, ...
    )
    streamable_http_app = StreamableHTTPASGIApp(session_manager)

    routes = []
    middleware = []

    # 认证中间件链
    if token_verifier:
        middleware = [
            Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(...)),
            Middleware(AuthContextMiddleware),
        ]
        routes.append(Route(
            streamable_http_path,
            endpoint=RequireAuthMiddleware(streamable_http_app, ...),
        ))
    else:
        routes.append(Route(streamable_http_path, endpoint=streamable_http_app))

    return Starlette(debug=debug, routes=routes, middleware=middleware,
                     lifespan=lambda app: session_manager.run())

这段代码展示了几个关键的架构特征：

SessionManager 与 Starlette lifespan 的绑定：通过 lifespan=lambda app: session_manager.run()，确保 session manager 的生命周期与 Starlette 应用一致。
可选的认证中间件链：通过条件判断决定是否添加 Bearer Token 认证。
DNS 重绑定防护：对 localhost 绑定自动启用安全防护。

10.5.3 自定义路由扩展

MCPServer 还支持通过 @custom_route 装饰器添加自定义 HTTP 路由：

@server.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> Response:
    return JSONResponse({"status": "ok"})

这些自定义路由不受 MCP 认证中间件保护，适合用于健康检查、OAuth 回调等公开端点。它们在路由列表中排在最后，确保 MCP 协议路由的优先级最高。

10.6 会话管理与初始化握手

10.6.1 ServerSession 的状态机

ServerSession 继承自 BaseSession，管理着一个三态状态机：

class InitializationState(Enum):
    NotInitialized = 1
    Initializing = 2
    Initialized = 3

状态转换的控制逻辑在 _received_request 中：

async def _received_request(self, responder):
    match responder.request:
        case types.InitializeRequest(params=params):
            self._initialization_state = InitializationState.Initializing
            self._client_params = params
            with responder:
                await responder.respond(types.InitializeResult(
                    protocol_version=...,
                    capabilities=self._init_options.capabilities,
                    server_info=types.Implementation(
                        name=self._init_options.server_name,
                        version=self._init_options.server_version,
                    ),
                ))
            self._initialization_state = InitializationState.Initialized
        case types.PingRequest():
            pass  # Ping 在任何状态下都允许
        case _:
            if self._initialization_state != InitializationState.Initialized:
                raise RuntimeError("Received request before initialization")

stateDiagram-v2
    [*] --> NotInitialized
    NotInitialized --> Initializing: 收到 InitializeRequest
    Initializing --> Initialized: 发送 InitializeResult
    NotInitialized --> NotInitialized: PingRequest (始终允许)
    Initializing --> Initializing: PingRequest (始终允许)
    Initialized --> Initialized: 处理所有请求/通知
    Initialized --> [*]: 连接关闭

    note right of NotInitialized
        除 Initialize 和 Ping 外的
        请求会抛出 RuntimeError
    end note

几个值得关注的设计细节：

Ping 例外：Ping 请求在任何状态下都被允许，这是 MCP 协议的要求，用于连接健康检查。
Stateless 模式：当 stateless=True 时，session 直接跳到 Initialized 状态，允许无状态 HTTP 场景下跳过握手。
客户端能力存储：_client_params 保存了客户端的初始化参数，后续可以通过 check_client_capability 查询客户端是否支持特定能力。

10.6.2 能力协商

ServerSession.check_client_capability 提供了细粒度的能力检查：

def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
    client_caps = self._client_params.capabilities

    if capability.roots is not None:
        if client_caps.roots is None:
            return False
        if capability.roots.list_changed and not client_caps.roots.list_changed:
            return False

    if capability.sampling is not None:
        if client_caps.sampling is None:
            return False
    ...
    return True

这使得工具函数可以在运行时根据客户端能力动态调整行为，例如只在客户端支持采样时才提供某些高级功能。

10.7 Context 注入机制

Context 是连接用户代码与 MCP 运行时的桥梁。它的注入机制基于类型注解的自动检测。

10.7.1 自动检测 Context 参数

find_context_parameter 函数扫描函数签名，查找类型注解为 Context 的参数：

# 用户只需声明参数类型为 Context
@server.tool()
async def my_tool(query: str, ctx: Context) -> str:
    await ctx.info(f"Processing: {query}")
    await ctx.report_progress(50, 100)
    result = await ctx.read_resource("data://source")
    return str(result)

SDK 会在注册时识别出 ctx 是 Context 参数，将其从 JSON Schema 生成中排除（客户端不应该传递 Context），并在调用时自动注入。

10.7.2 Context 提供的能力

Context 对象封装了丰富的运行时能力：

方法	用途
`ctx.info()` / `ctx.debug()` / `ctx.warning()` / `ctx.error()`	向客户端发送日志通知
`ctx.report_progress(progress, total, message)`	报告长任务进度
`ctx.read_resource(uri)`	读取其他注册的资源
`ctx.elicit(message, schema)`	向用户请求额外信息
`ctx.session`	访问底层 ServerSession
`ctx.request_id`	获取当前请求 ID

日志方法最终都委托给 ServerSession.send_log_message，这会向客户端发送 notifications/message 通知。进度报告则通过 ServerSession.send_progress_notification 发送 notifications/progress 通知。

10.8 生命周期管理（Lifespan）

Lifespan 机制允许在服务启动和关闭时执行初始化和清理逻辑，并将上下文数据传递给请求处理函数。

from contextlib import asynccontextmanager

@asynccontextmanager
async def app_lifespan(server: MCPServer):
    # 启动时：初始化资源
    db = await Database.connect("postgresql://localhost/mydb")
    redis = await Redis.connect("redis://localhost")
    try:
        yield {"db": db, "redis": redis}  # 传递给所有请求处理
    finally:
        # 关闭时：清理资源
        await redis.close()
        await db.close()

server = MCPServer("my-server", lifespan=app_lifespan)

@server.tool()
async def query(sql: str, ctx: Context) -> str:
    # 通过 ctx.request_context.lifespan_context 访问共享资源
    db = ctx.request_context.lifespan_context["db"]
    return await db.execute(sql)

Lifespan 在底层通过 MCPServer 到 Server 的包装函数传递：

def lifespan_wrapper(app, lifespan):
    @asynccontextmanager
    async def wrap(_: Server):
        async with lifespan(app) as context:
            yield context
    return wrap

这个包装的作用是将 MCPServer 实例（而非底层 Server 实例）传给用户的 lifespan 函数，使用户代码始终面向高层 API。

10.8.1 `LifespanResultT` 泛型：让 lifespan 上下文有类型

上面的例子把 lifespan 返回值当成 dict 用，每次 ctx.request_context.lifespan_context["db"] 访问一个字符串 key——写起来快，但没有类型检查。key 拼错要到运行时才知道。

真实的 MCPServer 是泛型类（server.py:129）：

class MCPServer(Generic[LifespanResultT]):
    def __init__(
        self,
        ...,
        lifespan: Callable[[MCPServer[LifespanResultT]],
                           AbstractAsyncContextManager[LifespanResultT]] | None = None,
        ...
    ):

LifespanResultT 是泛型参数——lifespan 返回什么类型、整个 MCPServer 实例就记住什么类型。配合 dataclass 可以写成：

from dataclasses import dataclass

@dataclass
class AppResources:
    db: Database
    redis: Redis

@asynccontextmanager
async def app_lifespan(server: MCPServer[AppResources]):
    db = await Database.connect("...")
    redis = await Redis.connect("...")
    try:
        yield AppResources(db=db, redis=redis)          # ← 结构化返回
    finally:
        await redis.close()
        await db.close()

server: MCPServer[AppResources] = MCPServer("my-server", lifespan=app_lifespan)

@server.tool()
async def query(sql: str, ctx: Context[AppResources, ...]) -> str:
    # 现在 ctx.request_context.lifespan_context 是 AppResources
    # IDE 自动补全 .db / .redis，拼错编译期抓住
    return await ctx.request_context.lifespan_context.db.execute(sql)

dict vs dataclass 的类型化选择是生产 MCP 服务器的常见分界线：demo 用 dict 够了；被多人维护、会改 lifespan 字段的真实服务强烈建议用 dataclass + 泛型标注——key 拼错 / 类型变更 / 字段删除都能静态检查。

10.8.2 `default_lifespan`：lifespan 的空实现

用户 MCPServer("foo") 不传 lifespan= 时，框架并不是”直接跳过整个机制”——而是替换成一个什么都不做的默认实现。打开 lowlevel/server.py:87：

@asynccontextmanager
async def lifespan(_: Server[LifespanResultT]) -> AsyncIterator[dict[str, Any]]:
    """Default lifespan context manager that does nothing.

    Returns:
        An empty context object
    """
    yield {}

关键设计决策：无论用户是否传 lifespan，底层 Server 总是以同一套调用协议处理——启动 → yield → 关闭。如果把”没传 lifespan”变成特殊分支（if lifespan is None: skip_context_setup()），整个 RequestContext 构造代码要到处分叉判断 lifespan_context is None。

用 default_lifespan 返回 {} 代替——调用方永远能拿到一个 lifespan_context、只是默认是空 dict。用值代替标志位——和前面 ch4/ch6/ch7 里 Serde tri!、Impossible、default_lifespan 同样的”消除特殊分支”思路。server.py:187 的三目选择器展示了这个替换：

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

10.8.3 `lifespan_wrapper` 里的 `_: Server` 类型错位

回头看 lifespan_wrapper（server.py:117）的真实签名：

def lifespan_wrapper(
    app: MCPServer[LifespanResultT],
    lifespan: Callable[[MCPServer[LifespanResultT]],
                       AbstractAsyncContextManager[LifespanResultT]],
) -> Callable[[Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]]:
    @asynccontextmanager
    async def wrap(_: Server[LifespanResultT]) -> AsyncIterator[LifespanResultT]:
        async with lifespan(app) as context:
            yield context
    return wrap

注意：

入参 lifespan 的签名要求接 MCPServer
返回的 wrap 函数签名要求接 Server（低层类型）
底层 Server.run() 会用 Server 实例调这个 wrap——但 wrap 收到后忽略它（_: Server）、改用 closure 里捕获的 app: MCPServer

这个类型 bending 是有意的：底层 Server.run 只认识 Server 类型签名、但用户的 lifespan 想面向 MCPServer 这个更丰富的 API 写。lifespan_wrapper 在两层类型之间做”适配器”——它接受低层类型、忽略它、替换成高层实例。

server.py:185-187 的源码注释明确说明这里存在 MCPServer 与低层 Server 的 lifespan 类型不匹配，并希望未来抽出一个像 Starlette 那样带 server 泛型的 Lifespan 类型。这个记录不是运行时 bug，而是 Python typing 在跨层适配时的真实折中：运行时闭包捕获的是高层实例，低层传入的参数被忽略；类型系统只能看到签名不完全一致，所以需要 type: ignore。对理解生产级 Python 库来说，这比“把类型擦干净”更有价值，因为它暴露了 API 易用性与静态类型完美性之间的取舍。

10.9 与 TypeScript SDK 的关键差异

维度	Python SDK	TypeScript SDK
架构分层	双层：MCPServer + Low-Level Server	单层：McpServer 直接包含所有逻辑
注册方式	装饰器 `@tool()` + Manager 类	`server.tool()` 方法注册回调
Schema 生成	Pydantic 从函数签名自动生成	Zod Schema 手动传入
异步框架	anyio（兼容 asyncio/trio）	原生 async/await
HTTP 框架	Starlette (ASGI)	内置 HTTP 处理
类型验证	Pydantic v2 运行时验证	TypeScript 编译时类型检查 + Zod 运行时验证
Context 注入	基于类型注解自动检测	显式传递
配置管理	pydantic-settings（支持环境变量）	构造函数参数

Python SDK 最大的优势在于 Pydantic 集成带来的自动 Schema 生成。开发者无需编写 JSON Schema 或 Zod Schema，只需要写 Python 类型注解即可。TypeScript SDK 则要求手动定义 Zod Schema 并传入 inputSchema。

另一方面，TypeScript SDK 的单层架构更加直观，而 Python SDK 的双层架构虽然灵活，但也增加了理解成本。

10.10 Settings 与环境变量配置

MCPServer 使用 pydantic-settings 实现配置管理，支持从环境变量和 .env 文件读取配置：

class Settings(BaseSettings, Generic[LifespanResultT]):
    model_config = SettingsConfigDict(
        env_prefix="MCP_",
        env_file=".env",
        env_nested_delimiter="__",
    )

    debug: bool
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
    warn_on_duplicate_resources: bool
    warn_on_duplicate_tools: bool
    warn_on_duplicate_prompts: bool
    dependencies: list[str]
    lifespan: Callable[...] | None
    auth: AuthSettings | None

通过 env_prefix="MCP_"，所有配置项都可以通过 MCP_ 前缀的环境变量设置。例如 MCP_DEBUG=true 会启用调试模式，MCP_LOG_LEVEL=DEBUG 会设置日志级别。嵌套配置通过 __ 分隔符支持，如 MCP_AUTH__ISSUER_URL=https://...。

这种设计使得 MCP 服务可以在不同环境（开发、测试、生产）中通过环境变量轻松调整行为，无需修改代码。

10.10.5 `mcp/server/` 目录的真实文件结构

mcp-python-sdk 的 src/mcp/server/ 目录实测包含——

文件 / 子目录	行	角色
`lowlevel/server.py`	672	底层 Server——纯 dispatcher
`mcpserver/server.py`	1112	高层 MCPServer——装饰器主体
`mcpserver/context.py`	273	Context 类（工具/资源回调的运行时入口）
`mcpserver/tools/base.py`	119	Tool 数据模型
`mcpserver/tools/tool_manager.py`	86	Tool 注册表 + dispatch
`mcpserver/resources/base.py`	44	Resource 基类
`mcpserver/resources/resource_manager.py`	108	Resource 注册表
`mcpserver/resources/templates.py`	133	URI 模板（`{var}` 占位）
`mcpserver/resources/types.py`	205	各类 Resource 实现（File/Function/HTTP/…）
`mcpserver/prompts/base.py`	189	Prompt 模型 + 渲染逻辑
`mcpserver/prompts/manager.py`	59	Prompt 注册表
`session.py`	-	会话生命周期
`streamable_http_manager.py`	-	Streamable HTTP transport
`sse.py` / `stdio.py` / `websocket.py`	-	三种基础 transport

双层架构的尺寸不对称——lowlevel/server.py 672 行 vs mcpserver/server.py 1112 行——高层比底层大近一倍。“皮肤”（装饰器、自动 schema 推断、参数验证、便捷方法）的代码量超过”骨架”（dispatcher）——这是大多数面向开发者的 SDK 的共性：易用性的代码代价是核心逻辑的 1.5-2 倍。

10.10.6 `MCPServer` 的 public API 完整清单

mcpserver/server.py:129 的 MCPServer 类——实测有 26 个公共方法——按用途分组：

元信息访问（7 个 property）——name / title / description / instructions / website_url / icons / version / session_manager

Tool 管理（3 个）——

add_tool(...) line 461 —— 命令式注册（传入函数对象）
remove_tool(name) line 501 —— 移除已注册工具
tool(...) line 512 —— 装饰器形式@server.tool()（用户最常用）

Resource 管理（2 个）——

add_resource(resource) line 619
resource(uri, ...) line 627 —— 装饰器@server.resource(uri="data://x")

Prompt 管理（2 个）——

add_prompt(prompt) line 739
prompt(...) line 747 —— 装饰器@server.prompt()

完成（1 个）——completion() line 582 —— @server.completion() 装饰自动补全 handler

自定义路由（1 个）——custom_route(...) line 805 —— 挂额外 ASGI 路由到 Starlette app（非 MCP 协议、用于 health check / metrics / OAuth callback 等）

Run 入口（多 overload）——run(transport=...) line 250-279——stdio / sse / streamable-http / websocket 四种 transport 选择

装饰器对偶 add_* 模式——三个核心原语都是”装饰器 + 命令式”双 API。装饰器适合静态注册、add_* 适合动态注册（运行时根据条件决定要不要加 tool）。这种”两路 API 等价”的设计常见于 Python SDK——比 TS SDK 的 single-style 更灵活。

10.10.7 `Context` 类——工具回调的”hidden 第一参数”

mcpserver/context.py:23 的 Context 类——MCP Python SDK 给工具函数提供的运行时能力入口：

class Context(BaseModel, Generic[LifespanContextT, RequestT]):
    """
    Tool/resource/prompt callable can declare a `ctx: Context` parameter
    to receive request context, progress reporting, logging, elicitation,
    and resource-reading capabilities.
    """

用户使用——

@server.tool()
async def my_tool(x: int, ctx: Context) -> str:
    await ctx.info(f"Processing {x}")
    await ctx.report_progress(0.5, total=1.0)
    data = await ctx.read_resource("data://config")
    return f"Done with {x}"

MCPServer 自动检测函数签名——如果有 ctx: Context 参数、调用时自动注入当前请求上下文；如果没有、就只传业务参数。用户写工具像写普通 async 函数、需要协议能力时声明一个 ctx 参数即可——零样板代码。

Context 提供 17+ 个方法——按 def/async def 签名实测——

类别	方法	行
进度	`report_progress(progress, total, message)`	87
资源读取	`read_resource(uri)`	108
用户互动	`elicit(...)` / `elicit_url(...)`	120 / 153
日志	`debug` / `info` / `warning` / `error`	259-271
通用 log	`log(...)`	188
元信息	`client_id` / `request_id` / `session`	212-222
SSE 流	`close_sse_stream()` / `close_standalone_sse_stream()`	226 / 243
导航	`mcp_server` / `request_context`（property）	74 / 81

这 17+ 个方法是 MCP 协议在工具执行期间所有能力的入口——不需要工具函数手动持有 session 引用、不需要传 progress callback——ctx 一统江湖。

对比 TS SDK——TS SDK 没有等价的 Context 单一对象——能力分散在 RequestHandlerExtra / ToolCallback 各种参数里——Python 的 ctx 是更优雅的封装。本书第 9 章 TS Client 的对比里、ctx 这种”capability bag”模式可以纳入”两个 SDK 设计哲学差异”的讨论。

10.10.8 `Tool.from_function` —— 自动 schema 推断的真身

用户写 @server.tool() 装饰 Python 函数后、MCP Server 怎么知道工具参数应该是 {"x": int, "y": str} 这样的 JSON Schema？答案在 mcpserver/tools/base.py:44 的 Tool.from_function：

@classmethod
def from_function(cls, fn, name=None, ...) -> Tool:
    func_name = name or fn.__name__
    func_doc = description or fn.__doc__ or ""
    is_async = is_async_callable(fn)
    context_kwarg = find_context_parameter(fn)   # ← 找 ctx: Context 参数

    func_arg_metadata = func_metadata(
        fn,
        skip_names=[context_kwarg] if context_kwarg else [],   # ← 排除 ctx
        structured_output=structured_output,
    )
    # 关键：把函数签名转成 Pydantic model、再用它生成 JSON Schema
    parameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True)

    return cls(fn=fn, name=func_name, ..., parameters=parameters, ...)

四步推断——

函数名 ← fn.__name__（用户没显式传 name 时）
描述 ← fn.__doc__（docstring 直接当工具描述给 LLM）
跳过 ctx ← find_context_parameter 自动识别 ctx: Context 参数、不出现在 JSON Schema 里
参数 schema ← func_metadata 用 inspect 读类型标注 + Pydantic 构造一个临时 BaseModel、再调 model_json_schema() 生成 OpenAPI-style JSON Schema

关键魔法在 func_metadata——它把 def my_tool(x: int, y: str = "hi") -> bool: 这种纯 Python 类型标注、变成等价的：

class MyToolArgs(BaseModel):
    x: int
    y: str = "hi"

然后 MyToolArgs.model_json_schema(by_alias=True) 输出标准 JSON Schema：

{
  "type": "object",
  "properties": {
    "x": {"type": "integer"},
    "y": {"type": "string", "default": "hi"}
  },
  "required": ["x"]
}

这就是 Pydantic 在 MCP Python SDK 的核心价值——用户用 Python 类型标注就够、不用学 JSON Schema。复杂场景（嵌套 dict、Optional、Annotated[Field]）也都自动处理。

对比 TS SDK——TS 用 zod 或 StandardSchemaWithJSON、用户必须显式传 schema（{ inputSchema: z.object({...}) }）——TS 类型不能 runtime 反射。Python 的 inspect + Pydantic 让”类型标注 = schema”、生态优势体现。

10.10.9 `ToolManager` 86 行核心

tool_manager.py 86 行——5 个核心方法：

class ToolManager:
    def __init__(self, warn_on_duplicate_tools=True, *, tools=None): ...
    def get_tool(self, name) -> Tool | None: ...      # O(1) dict lookup
    def list_tools(self) -> list[Tool]: ...           # 全量列表
    def add_tool(self, fn, ...) -> Tool: ...          # 注册（重名 warn 或 silent）
    def remove_tool(self, name) -> None: ...          # 注销

warn_on_duplicate_tools 参数——重复注册同名工具时——True 发 warning（默认）、False 静默替换。生产环境一般保持 True——能立即捕捉”两个文件不小心都注册了 read_file”这类 bug。

add_tool 内部委托给 Tool.from_function——把 schema 推断的复杂度集中在 Tool 类、ToolManager 只管”注册表 + dispatch”——单一职责。

整个 ToolManager 86 行——背后 86 行能撑起 MCP 协议 90% 的工具能力——是 Pydantic 生态厚度的红利。如果让 TS / Java 实现等价能力——估计要 300-500 行（手写 schema 验证器）。

10.10.10 从装饰器到协议 handler：Python Server 的真实调用链

把 Python Server 看作“几个装饰器”会低估它的分层。真实入口在 mcp-python-sdk/src/mcp/server/mcpserver/server.py:129-188：MCPServer 构造函数先创建 ToolManager、ResourceManager、PromptManager，再把 _handle_list_tools、_handle_call_tool、资源与 prompt handler 注入 low-level Server。这说明高层 API 负责收集 Python 函数和元数据，low-level Server 才负责 JSON-RPC method 到 handler 的分发。mcp-python-sdk/src/mcp/server/lowlevel/server.py:204-234 进一步验证了这一点：tools/list、tools/call、resources/read、completion/complete 等 method 都被收进 _request_handlers，而 roots/list_changed 与 progress 这类单向消息进入 _notification_handlers。

工具调用链也有清晰的两段式转换。server.py:512-580 的 tool() 装饰器只做一件事：把用户函数交给 add_tool() 注册，并原样返回函数；真正的函数签名分析在 mcp-python-sdk/src/mcp/server/mcpserver/tools/base.py:44-89，Tool.from_function() 会取函数名、docstring、异步性、Context 参数，并通过 func_metadata() 生成 JSON Schema。等到请求真的进来，server.py:302-333 的 _handle_call_tool() 会把底层 request context 包成高层 Context，然后调用 call_tool()；server.py:400-406 再转给 ToolManager.call_tool()；最后 tool_manager.py:74-86 找到对应 Tool，调用 Tool.run() 执行参数校验、Context 注入与结果转换。

资源和 prompt 走同一套“声明时轻、调用时重”的模式。server.py:627-737 的 resource() 会先检查 URI 模板参数与函数参数是否一致：如果 URI 包含 {city}，函数也必须有 city 参数，Context 参数会被排除在校验之外；否则就注册为普通 FunctionResource。server.py:747-803 的 prompt() 则把函数转成 Prompt 对象后交给 PromptManager。这样做的好处是，声明阶段尽早发现“URI 参数对不上函数签名”的错误，而调用阶段只需要按协议 method 路由，不再重复做结构判断。

传输入口也保持高低层分离。server.py:848-855 的 run_stdio_async() 只是取得 stdio 的读写流，然后调用 low-level server 的 run()；server.py:885-917 的 run_streamable_http_async() 构建 Starlette/uvicorn 应用；server.py:1045-1065 的 streamable_http_app() 又把 json_response、stateless_http、event_store、retry_interval、transport_security 和 auth 设置全部交给 low-level Server。也就是说，Python SDK 的高层 Server 不是协议实现本体，而是一层“Pythonic 声明式门面”：让用户用装饰器写业务函数，同时把协议 handler、生命周期、传输与认证留在下层组合。

这条链路解释了本章前面几个看似分散的设计：Context 为什么能作为隐藏参数注入，是因为 Tool.from_function() 先识别它，再由 Tool.run() 在调用时补进去；工具返回 dict 为什么能变成 structured content，是因为执行之后还有 convert_result；装饰器为什么必须写成 @server.tool() 而不是 @server.tool，是因为装饰器本身需要接收 name、title、annotations、structured_output 等注册配置。读懂这条链路，写 Python MCP Server 时就不会把“函数注册”“协议暴露”“请求执行”混成一件事。

这也能指导调试顺序：注册失败，先查装饰器是否真的被调用、ToolManager 里是否有名字；schema 不对，查 Tool.from_function() 识别到的签名、Context 参数是否被排除；调用失败，查 _handle_call_tool() 是否把异常转成 CallToolResult(is_error=True)，还是底层协议错误；传输不通，再看 stdio、SSE 或 Streamable HTTP 的 app 构建。按这条链路排查，比在用户函数里盲目 print 更快。

还有一个实践边界：高层 MCPServer 适合绝大多数业务 Server，但不是所有场景都必须从装饰器开始。如果你要动态生成工具、在运行时增删 handler、或把 MCP 嵌入已有 ASGI 应用，low-level Server 的 handler 字典更直接；如果你只是把 Python 函数暴露给模型，高层装饰器能少写大量 schema 和路由样板。两层并存不是重复设计，而是把“框架友好”和“协议可控”同时留给用户。

这种分层还影响测试。高层测试应围绕函数签名、装饰器注册、Context 注入和返回值转换；低层测试应围绕 JSON-RPC method、能力声明和 transport 输入输出。把两类测试混在一起，会导致一个工具函数改了类型注解，却让 transport 测试失败，或者一个 HTTP 头解析问题被误判成工具业务失败。

最稳妥的做法是给每个公开工具保留一组纯函数单元测试，再给 MCP 层保留一组协议级集成测试。前者保证业务逻辑正确，后者保证 schema、参数转换、错误包装和传输生命周期正确。

10.11 本章小结

本章深入剖析了 MCP Python SDK 服务端的核心实现。双层架构将易用性和灵活性完美分离，装饰器体系让工具注册回归函数签名的自然表达，Pydantic 集成消除了手写 Schema 的负担，anyio 提供了健壮的结构化并发模型，Starlette 集成则让 MCP 服务能够以标准 ASGI 应用的形式部署到任何生产级 ASGI 服务器上。

理解了这些实现细节之后，在下一章中，我们将切换到客户端视角，分析 Python SDK 的客户端实现，看看它如何与本章讨论的服务端配合工作。

第10章 Python Server 实现剖析

10.1 双层架构：MCPServer 与 Low-Level Server

10.2 装饰器体系：@tool、@resource、@prompt

10.2.1 @tool 装饰器

10.2.2 @resource 装饰器

10.2.3 @prompt 装饰器

10.3 Pydantic 深度集成

10.3.1 函数签名到 JSON Schema 的自动转换

10.3.2 参数验证与调用

10.3.3 Context 类的 Pydantic 基类设计

10.4 anyio 异步模型

10.4.1 为什么是 anyio

10.4.2 结构化并发在消息处理中的应用

10.4.3 stdio 传输的 anyio 实现

10.5 Starlette/ASGI 集成

10.5.1 多传输统一入口

10.5.2 Starlette 应用构建

10.5.3 自定义路由扩展

10.6 会话管理与初始化握手

10.6.1 ServerSession 的状态机

10.6.2 能力协商

10.7 Context 注入机制

10.7.1 自动检测 Context 参数

10.7.2 Context 提供的能力

10.8 生命周期管理（Lifespan）

10.8.1 LifespanResultT 泛型：让 lifespan 上下文有类型

10.8.2 default_lifespan：lifespan 的空实现

10.8.3 lifespan_wrapper 里的 _: Server 类型错位

10.9 与 TypeScript SDK 的关键差异

10.10 Settings 与环境变量配置

10.10.5 mcp/server/ 目录的真实文件结构

10.10.6 MCPServer 的 public API 完整清单

10.10.7 Context 类——工具回调的”hidden 第一参数”

10.10.8 Tool.from_function —— 自动 schema 推断的真身

10.10.9 ToolManager 86 行核心

10.10.10 从装饰器到协议 handler：Python Server 的真实调用链

10.11 本章小结

10.8.1 `LifespanResultT` 泛型：让 lifespan 上下文有类型

10.8.2 `default_lifespan`：lifespan 的空实现

10.8.3 `lifespan_wrapper` 里的 `_: Server` 类型错位

10.10.5 `mcp/server/` 目录的真实文件结构

10.10.6 `MCPServer` 的 public API 完整清单

10.10.7 `Context` 类——工具回调的”hidden 第一参数”

10.10.8 `Tool.from_function` —— 自动 schema 推断的真身

10.10.9 `ToolManager` 86 行核心