MCP 协议设计与实现
第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），提供基于装饰器的开发体验。这与第 8 章 TypeScript SDK 的低层 Server 加高层 McpServer 是同一种分层思路。
MCPServer 就是 v1 里的 FastMCP。本专栏 pin 的 3d7b311 处在 v1.25.0 之后、v2.0.0a1 之前的 v2 开发线上，仓库 docs/migration.md 把「FastMCP renamed to MCPServer」列为 v2 的破坏性变更：类改名，mcp.server.fastmcp.* 下的子模块原样迁到 mcp.server.mcpserver.*，pin 树的 src/ 里已经没有 fastmcp 模块。网上大量 from mcp.server.fastmcp import FastMCP 的示例，对应的就是本章的 from mcp.server.mcpserver import MCPServer。
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 保持极度简洁。
分发器也统一负责把 handler 的异常翻译成 JSON-RPC 错误（src/mcp/server/lowlevel/server.py:464-517）：handler 抛 MCPError 时回它携带的 ErrorData；抛其他任何异常时回 ErrorData(code=0, message=str(err))——code 是 0，不是 -32603；方法没有注册 handler 时回 -32601 Method not found。高层 MCPServer 里「未知资源」「未知 Prompt」抛的都是普通异常，所以实跑时线上看到的是：
<< {"jsonrpc":"2.0","id":5,"error":{"code":0,"message":"Unknown resource: config://missing"}}
<< {"jsonrpc":"2.0","id":6,"error":{"code":0,"message":"Unknown prompt: nope"}}工具不走这条路：MCPServer._handle_call_tool 先把非 MCPError 的异常（含未知工具、参数校验失败）接住，包成 isError: true 的 CallToolResult 正常返回（src/mcp/server/mcpserver/server.py:307-316）。
高层 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} 占位符，或函数带任何参数（包括只带 Context），都注册为模板资源
context_param = find_context_parameter(fn)
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 模板中的参数名与函数参数名是否一致，不一致则立即报错，而非等到运行时才发现问题。
注意判断条件是「URI 有占位符或函数有参数」。一个只声明了 ctx: Context 的资源函数，校验能通过（Context 被排除在比较之外），却会被注册成一个零变量的模板而不是静态资源：它出现在 resources/templates/list 里而不在 resources/list 里，实跑 resources/read 读它得到 code: 0, "Unknown resource: ..."（第 6 章 §6.6.2 有同一结论）。
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 decorator10.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 addArguments(ArgModelBase):<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（实跑输出，键序按 Pydantic 原样）：
{
"properties": {
"table": {"title": "Table", "type": "string"},
"limit": {"default": 10, "title": "Limit", "type": "integer"},
"filters": {
"anyOf": [
{"additionalProperties": {"type": "string"}, "type": "object"},
{"type": "null"}
],
"default": null,
"title": "Filters"
}
},
"required": ["table"],
"title": "query_databaseArguments",
"type": "object"
}每个属性的 title 和顶层的 query_databaseArguments 都是 Pydantic 自动加的——后者就是 func_metadata 动态创建的参数模型的类名（f"{func.__name__}Arguments"）。同一次注册还会顺带推导输出 schema：返回注解是 -> str 这类非对象类型时，func_metadata 把它包成 {"result": ...}，于是这个工具的 outputSchema 是 {"properties": {"result": {"title": "Result", "type": "string"}}, "required": ["result"], "title": "query_databaseOutput", "type": "object"}，调用结果除了文本 content 之外还会带上 structuredContent: {"result": ...}（实跑）。不想要结构化输出，就在装饰器里传 structured_output=False。
10.3.2 参数验证与调用
当工具被调用时，FuncMetadata.call_fn_with_arg_validation 会使用 Pydantic Model 对传入参数进行验证，然后再调用实际函数：
# src/mcp/server/mcpserver/tools/base.py:91（签名在源码里是多行带类型）
# 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 ePydantic 验证在这里把关：类型不匹配时会给出清晰的错误信息，而不是在函数内部产生难以追踪的运行时错误。校验失败同样落进 except Exception，被包成 ToolError，最终以 isError: true 的 CallToolResult 返回——实跑传 {"x": "abc"} 给 x: int 的工具，客户端收到的是 result.content[0].text 为 Error executing tool boom: 1 validation error for boomArguments ... 的正常响应，而不是 JSON-RPC 错误。
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 结构化并发的核心用法：
# Server.run，src/mcp/server/lowlevel/server.py:359（签名在源码里是多行带类型）
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, ...)
)
# 省略：experimental task support 的配置
async with anyio.create_task_group() as tg:
try:
async for message in session.incoming_messages:
if isinstance(message, RequestResponder) and message.context is not None:
context = message.context # 传输层捕获的发送方上下文
else:
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 上下文：请求优先在传输层随消息带来的发送方上下文里执行（例如 HTTP 请求处理协程里设置的 contextvar），没有时才用
contextvars.copy_context()复制当前上下文，确保每个处理任务拿到正确的 contextvar 值。 - 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 流式处理的典型模式：
# src/mcp/server/stdio.py 节选：省略 stdin/stdout 的 UTF-8 重新包装与 ClosedResourceError 处理
@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:
try:
message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
except Exception as exc:
await read_stream_writer.send(exc) # 坏行不终止循环，异常对象交给上层
continue
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(by_alias=True, exclude_unset=True)
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) 创建无缓冲的内存流，确保背压传导；这个包装还会在每次 send() 时快照发送方的 contextvars，供 Server.run 在接收端恢复（上一节的第 3 点）。第二，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 为例，MCPServer.streamable_http_app() 本身只是把参数连同 auth 配置、自定义路由一起转交给低层 Server.streamable_http_app()（src/mcp/server/mcpserver/server.py:1045-1070），真正构建 ASGI 应用的是低层这一份：
# Server.streamable_http_app，src/mcp/server/lowlevel/server.py:562-672（简化：
# 省略 OAuth 授权端点、受保护资源元数据端点的注册）
def streamable_http_app(self, *, streamable_http_path="/mcp", host="127.0.0.1", ...):
# localhost 绑定且未显式配置时，自动开启 DNS 重绑定防护
if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
transport_security = TransportSecuritySettings(enable_dns_rebinding_protection=True, ...)
session_manager = StreamableHTTPSessionManager(
app=self, event_store=event_store, ...
)
streamable_http_app = StreamableHTTPASGIApp(session_manager)
routes = []
middleware = []
# 认证中间件链（只在配置了 auth 且有 token_verifier 时装上）
if auth and token_verifier:
middleware = [
Middleware(AuthenticationMiddleware, backend=BearerAuthBackend(...)),
Middleware(AuthContextMiddleware),
]
if token_verifier:
routes.append(Route(
streamable_http_path,
endpoint=RequireAuthMiddleware(streamable_http_app, ...),
))
else:
routes.append(Route(streamable_http_path, endpoint=streamable_http_app))
if custom_starlette_routes:
routes.extend(custom_starlette_routes) # @custom_route 注册的路由排在最后
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 重绑定防护：
host是127.0.0.1/localhost/::1且没传transport_security时，自动只放行本机的 Host 与 Origin。
10.5.3 自定义路由扩展
MCPServer 还支持通过 @custom_route 装饰器添加自定义 HTTP 路由：
@server.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> Response:
return JSONResponse({"status": "ok"})这些自定义路由不经过 RequireAuthMiddleware 包装、不要求认证，适合用于健康检查、OAuth 回调等公开端点。它们在路由列表中排在最后，确保 MCP 协议路由的优先级最高。
10.6 会话管理与初始化握手
10.6.1 ServerSession 的状态机
ServerSession 继承自 BaseSession，管理着一个三态状态机：
class InitializationState(Enum):
NotInitialized = 1
Initializing = 2
Initialized = 3状态转换的控制逻辑在 _received_request 中：
# ServerSession._received_request，src/mcp/server/session.py:165
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 was complete")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，
客户端收到 -32602 Invalid request parameters
end note
有两处行为要看清。第一，状态在回复完 InitializeResult 之后就置为 Initialized，并不等客户端的 notifications/initialized——那条通知到达时 _received_notification 只是再把状态设一次 Initialized（src/mcp/server/session.py:165-205）。实跑时客户端收到 initialize 响应后不发 initialized 通知、直接发 tools/call，服务端照常处理（第 4 章有同一结论）。第二，RuntimeError 并不会让连接崩掉：它在 BaseSession._receive_loop 里被一个笼统的 except Exception 接住，回给客户端的是 -32602 Invalid request parameters（src/mcp/shared/session.py:377-388），报文里看不出「还没初始化」这层原因：
>> {"jsonrpc": "2.0", "id": 0, "method": "tools/list"}
<< {"jsonrpc":"2.0","id":0,"error":{"code":-32602,"message":"Invalid request parameters","data":""}}几个值得关注的设计细节：
- Ping 例外：Ping 请求在任何状态下都被放行。规范（2025-11-25
basic/lifecycle.mdx）规定服务端回复 initialize 之前，客户端除 ping 外 SHOULD NOT 发其他请求，所以 ping 必须能在未初始化状态下得到响应，用于连接健康检查。 - Stateless 模式：当
stateless=True时，session 直接跳到Initialized状态，允许无状态 HTTP 场景下跳过握手。 - 客户端能力存储：
_client_params保存了客户端的初始化参数，后续可以通过check_client_capability查询客户端是否支持特定能力。
10.6.2 能力协商
ServerSession.check_client_capability 提供了细粒度的能力检查：
# src/mcp/server/session.py:122
def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
if self._client_params is None: # 还没收到 initialize
return False
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 是泛型类（src/mcp/server/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
from typing import Any
@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, Any]) -> 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= 时，框架并不是"直接跳过整个机制"——而是替换成一个什么都不做的默认实现。打开 src/mcp/server/lowlevel/server.py:87（它在低层模块里就叫 lifespan，src/mcp/server/mcpserver/server.py:32 以 default_lifespan 的名字导入）：
@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。这就是用「空值对象」代替标志位、从而消除特殊分支的经典思路。src/mcp/server/mcpserver/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（src/mcp/server/mcpserver/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 在两层类型之间做"适配器"——它接受低层类型、忽略它、替换成高层实例。
src/mcp/server/mcpserver/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 组合低层 Server（pin 版低层 Server 标了 @deprecated，只建议高级场景直接用） |
| 注册方式 | 装饰器 @tool() + Manager 类 |
registerTool() / registerResource() / registerPrompt() 方法注册回调（v1 的 server.tool() 等变参重载在 pin 版已删除） |
| Schema 生成 | Pydantic 从函数签名自动生成 | Standard Schema（如 Zod）手动传入 |
| 异步框架 | anyio（兼容 asyncio/trio） | 原生 async/await |
| HTTP 框架 | Starlette (ASGI) | 基于 Web 标准 Request/Response 的传输，另有 Node / Express / Hono 适配包 |
| 类型验证 | Pydantic v2 运行时验证 | TypeScript 编译时类型检查 + Standard Schema 运行时验证 |
| Context 注入 | 基于类型注解自动检测 | 作为 handler 的参数显式传入 |
| 配置管理 | 构造函数参数（内部存进 pydantic-settings 的 Settings，但环境变量实际不生效，见 §10.10） |
构造函数参数 |
Python SDK 最大的优势在于 Pydantic 集成带来的自动 Schema 生成。开发者无需编写 JSON Schema 或 Zod Schema，只需要写 Python 类型注解即可。TypeScript SDK 则要求手动定义 Standard Schema 对象（如 z.object({...})）并传入 inputSchema。
分层上两者并无差别：都是高层门面负责易用性、低层 Server 负责协议分发。
10.10 Settings 与环境变量配置
MCPServer 把构造参数存进一个基于 pydantic-settings 的 Settings 对象：
# src/mcp/server/mcpserver/server.py:80
class Settings(BaseSettings, Generic[LifespanResultT]):
model_config = SettingsConfigDict(
env_prefix="MCP_",
env_file=".env",
env_nested_delimiter="__",
nested_model_default_partial_update=True,
extra="ignore",
)
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看配置和类的 docstring（「All settings can be configured via environment variables with the prefix MCP_」），似乎 MCP_DEBUG=true、MCP_LOG_LEVEL=DEBUG、MCP_AUTH__ISSUER_URL=... 都能生效。但实现不是这样：MCPServer.__init__ 把每个字段都作为关键字参数显式传给 Settings(...)（src/mcp/server/mcpserver/server.py:153-162），而 pydantic-settings 里初始化参数的优先级高于环境变量和 .env，于是构造函数的默认值（debug=False、log_level="INFO"、auth=None……）总是赢。实跑 MCP_DEBUG=true MCP_LOG_LEVEL=DEBUG 启动后，server.settings.debug 仍是 False、log_level 仍是 "INFO"；放一个写了 MCP_DEBUG=true 的 .env、设 MCP_AUTH__ISSUER_URL 也一样不生效。要按环境切换配置，得自己读环境变量再传给 MCPServer(...)。
10.10.5 mcp/server/ 目录的真实文件结构
mcp-python-sdk 的 src/mcp/server/ 目录实测包含（节选，行数为 wc -l）——
| 文件 / 子目录 | 行 | 角色 |
|---|---|---|
src/mcp/server/lowlevel/server.py |
672 | 底层 Server——纯 dispatcher |
src/mcp/server/mcpserver/server.py |
1112 | 高层 MCPServer——装饰器主体 |
src/mcp/server/mcpserver/context.py |
273 | Context 类（工具/资源回调的运行时入口） |
src/mcp/server/mcpserver/tools/base.py |
119 | Tool 数据模型 |
src/mcp/server/mcpserver/tools/tool_manager.py |
86 | Tool 注册表 + dispatch |
src/mcp/server/mcpserver/resources/base.py |
44 | Resource 基类 |
src/mcp/server/mcpserver/resources/resource_manager.py |
108 | Resource 注册表 |
src/mcp/server/mcpserver/resources/templates.py |
133 | URI 模板（{var} 占位） |
src/mcp/server/mcpserver/resources/types.py |
205 | 各类 Resource 实现（File/Function/HTTP/...） |
src/mcp/server/mcpserver/prompts/base.py |
189 | Prompt 模型 + 渲染逻辑 |
src/mcp/server/mcpserver/prompts/manager.py |
59 | Prompt 注册表 |
src/mcp/server/session.py |
692 | ServerSession——初始化状态机与会话 |
src/mcp/server/streamable_http.py |
1056 | Streamable HTTP transport |
src/mcp/server/streamable_http_manager.py |
296 | Streamable HTTP 会话管理（StreamableHTTPSessionManager） |
src/mcp/server/sse.py / stdio.py / websocket.py |
241 / 77 / 52 | 另外三种 transport |
双层架构的尺寸不对称——src/mcp/server/lowlevel/server.py 672 行 vs src/mcp/server/mcpserver/server.py 1112 行——高层比底层大不少。"皮肤"（装饰器、自动 schema 推断、参数验证、便捷方法）的代码量超过"骨架"（dispatcher），这在面向开发者的 SDK 里并不少见：易用性本身是要用代码换的。
10.10.6 MCPServer 的 public API 完整清单
src/mcp/server/mcpserver/server.py:129 的 MCPServer 类——实测有 8 个 property + 22 个公共方法——按用途分组：
元信息访问（8 个 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 619resource(uri, ...)line 627 —— 装饰器@server.resource(uri="data://x")
Prompt 管理（2 个）——
add_prompt(prompt)line 739prompt(...)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 249-300——stdio / sse / streamable-http 三种 transport 选择（run() 的 match 分支只认这三个字面量；websocket 有独立的 websocket_server() 传输实现，但不在 MCPServer.run() 的选项里）
装饰器对偶 add_* 模式——三个核心原语都是"装饰器 + 命令式"双 API。装饰器适合静态注册、add_* 适合动态注册（运行时根据条件决定要不要加 tool）。TS SDK 只有 registerTool() 这一种命令式写法，动态注册同样靠它。
10.10.7 Context 类——工具回调的"hidden 第一参数"
src/mcp/server/mcpserver/context.py:23 的 Context 类——MCP Python SDK 给工具函数提供的运行时能力入口：
class Context(BaseModel, Generic[LifespanContextT, RequestT]):
"""Context object providing access to MCP capabilities.
This provides a cleaner interface to MCP's RequestContext functionality.
It gets injected into tool and resource functions that request it via type hints.
...
The context parameter name can be anything as long as it's annotated with Context.
The context is optional - tools that don't need it can omit the parameter.
"""用户使用——
@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 提供 16 个公开成员（11 个方法 + 5 个 property）——按 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 |
这 16 个成员是 MCP 协议在工具执行期间所有能力的入口——不需要工具函数手动持有 session 引用、不需要传 progress callback——ctx 一统江湖。
对比 TS SDK——这里值得讲的不是"谁更优雅"，而是两个 SDK 独立收敛到了同一个设计。
TS SDK 1.x 的 handler 第二个参数是 RequestHandlerExtra，signal、requestId、sendNotification、authInfo 等字段全部平铺在这一个对象上。2.0 把它重组成分组的 ServerContext（packages/core/src/shared/protocol.ts）：
export type ServerContext = BaseContext & {
mcpReq: {
log: (level, data, logger?) => Promise<void>; // 发日志通知，遵守 client 的 level 过滤
elicitInput: (params, options?) => Promise<ElicitResult>; // 向用户要输入
// …sampling 等
};
http?: { authInfo?: AuthInfo }; // 仅 HTTP 传输时存在
task?: TaskContext; // 配置了 task 存储时存在
}; // 简化：authInfo 与 task 实际定义在 BaseContext 里，ServerContext 的 http 再补上 req、closeSSE 等Server 的类型签名就是 class Server extends Protocol<ServerContext>，每个 handler 拿到的第二个参数即是它——和 Python 的 ctx 是同一个模式：把协议在请求期间的全部能力打成一个包交给处理函数。
RequestHandlerExtra 这个名字在 pin 版的源码里已经不存在，只剩 docs/migration.md 的 v1→v2 对照表还把它列为待替换的旧类型名（extra.authInfo → ctx.http?.authInfo、extra.taskStore → ctx.task?.store 等）。
真正的差异在可选性上：Python 的 ctx 是一个对象，能力齐备；TS 的 ServerContext 用 http? / task? 把"只在某种传输或某种配置下才存在"的能力显式标成可选，让类型系统替你挡住"在 stdio 下访问 authInfo"这类错误。两种取向各有代价——一个读起来更顺，一个编译期更严。
10.10.8 Tool.from_function —— 自动 schema 推断的真身
用户写 @server.tool() 装饰 Python 函数后、MCP Server 怎么知道工具参数应该是 {"x": int, "y": str} 这样的 JSON Schema？答案在 src/mcp/server/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()生成 JSON Schema
关键魔法在 func_metadata——它把 def my_tool(x: int, y: str = "hi") -> bool: 这种纯 Python 类型标注、变成等价的：
class my_toolArguments(BaseModel): # 实际基类是 ArgModelBase，类名为 f"{fn.__name__}Arguments"
x: int
y: str = "hi"然后 my_toolArguments.model_json_schema(by_alias=True) 输出标准 JSON Schema（实跑）：
{
"properties": {
"x": {"title": "X", "type": "integer"},
"y": {"default": "hi", "title": "Y", "type": "string"}
},
"required": ["x"],
"title": "my_toolArguments",
"type": "object"
}这就是 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: ... # 注册（重名时保留旧的）
def remove_tool(self, name) -> None: ... # 注销（不存在则抛 ToolError）
async def call_tool(self, name, arguments, context, convert_result=False): ... # dispatchwarn_on_duplicate_tools 参数——重复注册同名工具时——注意 SDK 的实际行为是保留先注册的那个并把它原样返回，新工具被丢弃；True（默认）额外打一条 Tool already exists: {name} 的 warning，False 则静默丢弃。生产环境一般保持 True——能立即捕捉"两个文件不小心都注册了 read_file"这类 bug。
add_tool 内部委托给 Tool.from_function——把 schema 推断的复杂度集中在 Tool 类、ToolManager 只管"注册表 + dispatch"——单一职责。
整个 ToolManager 只有 86 行——因为真正的重活（签名解析、schema 生成、参数校验）都由 Pydantic 和 Tool.from_function 扛走了，这是 Pydantic 生态厚度的红利。
10.10.10 从装饰器到协议 handler：Python Server 的真实调用链
把 Python Server 看作“几个装饰器”会低估它的分层。真实入口在 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 的分发。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。
工具调用链也有清晰的两段式转换。src/mcp/server/mcpserver/server.py:512-580 的 tool() 装饰器只做一件事：把用户函数交给 add_tool() 注册，并原样返回函数；真正的函数签名分析在 src/mcp/server/mcpserver/tools/base.py:44-89，Tool.from_function() 会取函数名、docstring、异步性、Context 参数，并通过 func_metadata() 生成 JSON Schema。等到请求真的进来，src/mcp/server/mcpserver/server.py:307-333 的 _handle_call_tool() 会把底层 request context 包成高层 Context，然后调用 call_tool()；src/mcp/server/mcpserver/server.py:400-406 再转给 ToolManager.call_tool()；最后 src/mcp/server/mcpserver/tools/tool_manager.py:74-86 找到对应 Tool，调用 Tool.run() 执行参数校验、Context 注入与结果转换。
资源和 prompt 走同一套“声明时轻、调用时重”的模式。src/mcp/server/mcpserver/server.py:627-737 的 resource() 会先检查 URI 模板参数与函数参数是否一致：如果 URI 包含 {city}，函数也必须有 city 参数，Context 参数会被排除在校验之外；只有 URI 和函数都不带参数时，才注册为普通 FunctionResource。src/mcp/server/mcpserver/server.py:747-803 的 prompt() 则把函数转成 Prompt 对象后交给 PromptManager。这样做的好处是，声明阶段尽早发现“URI 参数对不上函数签名”的错误，而调用阶段只需要按协议 method 路由，不再重复做结构判断。
传输入口也保持高低层分离。src/mcp/server/mcpserver/server.py:848-855 的 run_stdio_async() 只是取得 stdio 的读写流，然后调用 low-level server 的 run()；src/mcp/server/mcpserver/server.py:885-917 的 run_streamable_http_async() 构建 Starlette/uvicorn 应用；src/mcp/server/mcpserver/server.py:1045-1070 的 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() 在调用时补进去；工具返回值为什么能带上 structured content，是因为注册时 func_metadata 按返回注解推导了输出模型，执行之后再由 convert_result 按它校验并生成 structuredContent；装饰器为什么必须写成 @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，但不是所有场景都必须从装饰器开始。如果你要每次 tools/list 都现算工具列表、自己构造完整的 CallToolResult，或者需要高层没有暴露的方法（如 resources/subscribe），low-level Server 的 on_* 回调更直接；如果你只是把 Python 函数暴露给模型，高层装饰器能少写大量 schema 和路由样板。两层并存不是重复设计，而是把“框架友好”和“协议可控”同时留给用户。
10.11 本章小结
本章深入剖析了 MCP Python SDK 服务端的核心实现。双层架构把易用性和灵活性分在两层，装饰器体系让工具注册回归函数签名的自然表达，Pydantic 集成消除了手写 Schema 的负担，anyio 提供了健壮的结构化并发模型，Starlette 集成则让 MCP 服务能够以标准 ASGI 应用的形式部署到任何生产级 ASGI 服务器上。
理解了这些实现细节之后，在下一章中，我们将切换到客户端视角，分析 Python SDK 的客户端实现，看看它如何与本章讨论的服务端配合工作。