杨艺韬讲堂

Appearance

第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 类形成鲜明对比。

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

python
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：

python
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() 装饰器的实现非常精巧：

python
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 完成函数到工具的转换：

python
@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, ...)

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

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

10.2.2 @resource 装饰器 ​

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

python
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

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

python
# 静态资源 —— 函数无参数
@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 从函数签名中提取参数信息：

python
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。

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

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

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

json
{
  "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 对传入参数进行验证，然后再调用实际函数：

python
# 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：

python
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 结构化并发的核心用法：

python
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()

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

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

10.4.3 stdio 传输的 anyio 实现 ​

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

python
@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() 是同步入口方法，它根据传输类型分派到不同的异步实现：

python
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 应用：

python
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())

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

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

10.5.3 自定义路由扩展 ​

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

python
@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，管理着一个三态状态机：

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

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

python
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")

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

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

10.6.2 能力协商 ​

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

python
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 的参数：

python
# 用户只需声明参数类型为 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 机制允许在服务启动和关闭时执行初始化和清理逻辑，并将上下文数据传递给请求处理函数。

python
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 的包装函数传递：

python
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.9 与 TypeScript SDK 的关键差异 ​

维度Python SDKTypeScript 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 文件读取配置：

python
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.11 本章小结 ​

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

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

基于 VitePress 构建

Copyright © 2024-present 杨艺韬