MCP Python SDK 协议层实现机制详解

BaseSession 类中的泛型参数：

首先我们看一下类中包含一系列的泛型参数（Generic）。这些类型参数一方面让方法签名更严谨（例如 send_request(request: SendRequestT)），另一方面在收到消息时将原始 JSONRPC 转成具体的 Pydantic 模型（ReceiveRequestT/ReceiveNotificationT），方便上层直接拿到结构化数据。

Generic[
    SendRequestT,       # 我们能发出的 Request 类型
    SendNotificationT,  # 我们能发出的 Notification 类型
    SendResultT,        # 我们发出的 Response（Result）类型
    ReceiveRequestT,    # 我们会接收的 Request 类型
    ReceiveNotificationT, # 我们会接收的 Notification 类型
    ReceiveResultT      # 我们会接收的 Response（Result）类型
]

BaseSession 类的核心属性和上下文管理

BaseSession 类的核心属性包括：

• _read_stream / _write_stream：AnyIO 内存流，用于异步收发 SessionMessage（JSONRPC 封装 + 元数据）。
• _request_id：内部自增计数器，用来给每个发出的请求分配唯一 ID。
• _response_streams: dict[RequestId, MemoryObjectSendStream]：每发一个请求就开一个答案流，收到对应 ID 的响应或错误时就往流里写，send_request 在此流上等待结果。
• _in_flight: dict[RequestId, RequestResponder]：跟踪当前正在处理但尚未完成的对端请求，方便在收到取消通知时进行清理或超时管理。
• _exit_stack：AnyIO 的异步上下文管理栈，用来统一关闭流和任务。BaseSession 类通过 __aenter__ / __aexit__ 进行上下文管理。
• aenter：创建一个 TaskGroup 并启动 _receive_loop，开始并行地监听对端发来的所有消息。
• aexit：依次关闭所有资源、取消接收循环任务，保证退出时不会挂起。

三、BaseSession 类中的基本方法

BaseSession 类中有一系列协议实现方法，主要包括 send_request()、send_notification()、_send_response() 和 _receive_loop() 等，用于控制 MCP 基本交互流程。

发送请求：send_request() 发送请求方法完成的是如何把一个高层的 Pydantic 请求模型打包成 JSON-RPC 消息、发出去，并在底层管理好"谁发的请求""等待谁的回应"这件事。

具体过程如下：

1. 分配请求 ID：用 _request_id 自增取新 ID。
2. 创建响应流：在 _response_streams 中为该 ID 注册一个缓冲大小为 1 的内存流。
3. 构造 JSONRPCRequest：将 Pydantic 模型 request 序列化为 JSONRPC 格式，附带 ID。
4. 写入 _write_stream：发往对端。
5. 等待响应：在对应的响应流上阻塞读取，设置超时（可自定义或继承会话级别超时）。
6. 错误处理：若拿到 JSONRPCError 就抛 McpError；若超时就抛带有 HTTP 408 码的 McpError；否则把 result 部分反序列化为 ReceiveResultT 并返回。
7. 清理：无论成功或失败，都要把流和注册记录删除并关闭。

发送响应：_send_response()

当收到对端的请求时，如何基于成功或失败结果，及时生成正确的 JSON-RPC 响应或错误，并写回去呢？

当收到对端发来的 Request，需要给它回复时：

• 如果是正常结果 SendResultT，就构造 JSONRPCResponse（含 ID + result 字段）；
• 如果是错误 ErrorData，就构造 JSONRPCError；
• 然后同样写入 _write_stream。

发送通知：send_notification()

在 send_notification() 方法中，会将那些只需单向推送、不需要回复的事件（如日志、进度、资源变化）等静悄悄地发给对端而不阻塞。具体实现上，是在方法内部构造 JSONRPCNotification（无 ID），部分传输层可携带 related_request_id（比如把进度通知关联到某次调用）；同时直接写入 _write_stream，不等待任何响应。

消息接收循环：_receive_loop()

_receive_loop() 方法会不断从底层流里读入任意一方发来的 JSON-RPC 消息，按类型分流到"请求处理""通知处理"或"回应分发"的正确管道里。

该方法会持续读取 _read_stream 中的 SessionMessage，并根据消息类型分流。

1. JSONRPCRequest：用 ReceiveRequestT 模型校验后，包装成 RequestResponder 放入 _in_flight，先调用 _received_request(responder) 钩子，让子类有机会同步响应；如果钩子未完成响应，再把 responder 传给 _handle_incoming（供上层或框架线程处理）。
2. JSONRPCNotification：用 ReceiveNotificationT 校验后，调用 _received_notification(notification) 钩子，然后 _handle_incoming。如果是取消通知，还会查找并取消对应的 in-flight 请求。
3. JSONRPCResponse / JSONRPCError：根据消息中的 ID，查找 _response_streams，并将结果或错误写入那个流，让 send_request 能读到。若找不到对应流，则视为意外消息，转交 _handle_incoming 处理。

可重写的钩子方法

BaseSession 类中还定义了一系列的可重用钩子方法，将上面分流后的"该如何处理"留给子类去实现——比如 ClientSession、ServerSession 用于握手、工具调用分发、能力校验等。

• _received_request(self, responder)
• _received_notification(self, notification)
• _handle_incoming(self, req_or_notification_or_exception)

BaseSession 的子类（如 ServerSession、ClientSession）会重写这些方法，注入初始化握手逻辑、工具调用分发、事件转发等具体行为，而无需再触及底层 JSON-RPC 和并发细节。通过以上这些模块化、泛型化和异步化的设计，MCP 在保证性能和灵活性的同时，也极大降低了开发者对底层 JSON-RPC 细节的认知成本。

四、ClientSession，ServerSession 类的上层实现

BaseSession 将 JSON-RPC 的消息帧化、ID 跟踪、并发读写、超时与错误处理都做好了。上层只需继承并重写少数钩子，就能快速实现"MCP 客户端"或"MCP 服务端"的会话逻辑。

我们再来看 client/session.py 和 server/session.py 两个文件，是如何"填充"这一骨架，把 MCP 协议真正落地成可用的客户端与服务端逻辑的。

客户端 (ClientSession) 如何"完成"协议设计

在 ClientSession 中，初始化阶段通过构造函数将下游的读写流、超时设置以及各类回调（包括取样、列出根目录、日志和通用消息处理）注入会话实例。调用 initialize() 时，客户端首先向服务端发送一个 initialize 请求，并在收到 InitializeResult 后，再发出一条 InitializedNotification 通知，让握手完成。

return await self.send_request(
    types.ClientRequest(types.ListResourcesRequest(method="resources/list")),
    types.ListResourcesResult,
)

在消息接收方面，ClientSession 重写了 _received_request(responder) 钩子，当服务端发来诸如 sampling/createMessage、roots/list 或 ping 等请求时，钩子会从 responder.request.root 中提取请求类型和参数，调用相应的回调（如 _sampling_callback 或 _list_roots_callback），并将返回值（可能是正常结果或 ErrorData）传给 responder.respond() 完成响应。

对于通知，_received_notification(notification) 目前只对日志消息（LoggingMessageNotification）作出处理，其他非同步钩子捕获的消息和未在钩子中处理的所有请求、通知，都会统一委托给用户在构造时提供的 message_handler，以便在上层进行监控、界面更新或其他自定义处理。

服务端 (ServerSession) 如何"完成"协议设计

在 ServerSession 中，协议的握手逻辑被内嵌在 _received_request 和 _received_notification 两个钩子里：首次碰到来自客户端的 InitializeRequest，服务器会将请求参数保存到 self._client_params，随即利用 responder.respond(…) 返回包含服务端能力集、版本号和说明文字的 InitializeResult。当接收到 InitializedNotification 时，状态机才真正进入"Initialized"阶段，任何在未完成初始化前到来的其他请求或通知都会触发 RuntimeError，以确保交互顺序的正确性。

握手完成后，check_client_capability(cap) 方法会读取客户端在初始化时声明的能力集，并根据传入的能力需求判断哪些功能可用——例如是否允许发布 roots/list_changed 等通知，从而在后续工具调用中进行权限或兼容性检查。

为了让服务端调用体验更自然，ServerSession 提供了一系列高层接口：诸如 send_log_message(…)、send_resource_updated(uri)、send_tool_list_changed() 之类的方法，它们均转换成对应的 ServerNotification 并通过底层 send_notification 发送；同样地，create_message(…)、list_roots()、send_ping() 等方法封装了对 JSON-RPC 请求的构建与发送，让服务器可以像调用本地函数一样驱动模型取样、根列表查询或心跳检测。

上行消息的分发，ServerSession 在 _handle_incoming 中会将所有未由同步钩子处理的 RequestResponder 或 Notification 推送到自身维护的内存队列中，这正是框架层面通过 @server.call_tool()、@server.list_prompts() 等装饰器将队列内的请求与通知自动路由到用户业务函数的基础。

由此，ServerSession 不仅完成了底层 JSON-RPC 交互的所有复杂细节，也为上层的工具调用和事件处理提供了极简且一致的编程模型。

从"通用"到"可用"的核心流程

在 MCP 的典型部署中，首先由 Host 应用在启动时分别实例化 ClientSession 和 ServerSession，并将二者的底层读写流对接：本地场景通常将一个进程的标准输出连向另一个进程的标准输入，而远程场景则可以通过 HTTP 与 SSE 通道互联。

接着，客户端调用 initialize()，通过底层 send_request 向服务端发出 initialize 请求；与此同时，ServerSession 在其接收钩子中捕获到该请求并立即返回包含能力集与版本信息的 InitializeResult，客户端再发出一条 InitializedNotification 以完成握手，这样双方就进入了"已初始化"阶段，准备开始正常的消息交换。

在交互过程中，任何一方都只需像调用本地函数一样发起高层 API 调用。例如，客户端只要执行 await session.list_tools()，ClientSession 就会在内部构造一个 JSON-RPC 请求，将其通过 send_request 发出，并在 ServerSession 的 _receive_loop 中被接收，随后被封装成一个 RequestResponder 并交给注册在服务器端的工具实现函数处理；工具函数执行完成后调用 responder.respond()，ServerSession 底层的 _send_response 将结果打包回传，最后在客户端的 send_request 中被捕获并反序列化成 ListToolsResult 返回给调用者。

同样地，服务端也可以直接调用诸如 await server_session.create_message(…) 来触发大模型，并将结果通过响应或单向通知发回客户端，为大模型的上下文生成提供支持。

上层只要在 ServerSession 或 ClientSession 外面注册回调，就能无缝插入新的工具、扩展新的消息类型或自定义错误处理，BaseSession 的任何改动都不影响业务逻辑。

可以看到，client/session.py 和 server/session.py 这两个模块，在 BaseSession 打造的坚固底座上，完成了握手流程、能力协商、消息分发、工具调用、高层接口封装，以及上下文管理，真正将 MCP 从"一份协议文档"变成了"开箱即用、可扩展的客户端 / 服务端 SDK"。

总结

Session 类是 MCP SDK 在底层对"JSON-RPC 消息收发"的统一封装——它由框架内部创建和管理。如我们之前用过的 FastMCP("weather") 其实在背后就会实例化一个 Session，并把它绑定到你的工具（@mcp.tool）以及资源发现接口上。

在编写 Server 代码时，只需关注声明工具函数、调用 mcp.run(…) 启动即可——不需要也看不到 Session 的具体实现。当想做非常定制化的"底层消息拦截"、"自定义请求路由"或"替换 JSON-RPC 实现"时，才需要考虑继承或修改 Session。