MCP 协议层详解：深入分析 Python SDK 中的实现机制

核心属性与上下文管理

BaseSession 的核心属性包括：

• _read_stream / _write_stream：AnyIO 内存流，用于异步收发 SessionMessage。
• _request_id：内部自增计数器，为每个发出的请求分配唯一 ID。
• _response_streams：字典结构，每发一个请求就开一个答案流，收到对应 ID 的响应或错误时写入，send_request 在此流上等待结果。
• _in_flight：跟踪当前正在处理但尚未完成的对端请求，便于取消通知时的清理或超时管理。
• _exit_stack：AnyIO 的异步上下文管理栈，统一关闭流和任务。

通过 __aenter__ / __aexit__ 进行上下文管理。进入时创建 TaskGroup 并启动 _receive_loop，并行监听对端消息；退出时依次关闭所有资源、取消接收循环任务。

三、BaseSession 类中的基本方法

BaseSession 提供了一系列协议实现方法，控制 MCP 基本交互流程。

发送请求：send_request()

该方法将高层 Pydantic 请求模型打包成 JSON-RPC 消息并发出，同时管理'谁发的请求'和'等待谁的回应'。具体过程如下：

1. 分配请求 ID：使用 _request_id 自增获取新 ID。
2. 创建响应流：在 _response_streams 中为该 ID 注册缓冲大小为 1 的内存流。
3. 构造 JSONRPCRequest：序列化 Pydantic 模型，附带 ID。
4. 写入流并等待：发往对端，在对应的响应流上阻塞读取，设置超时。
5. 错误处理：若拿到 JSONRPCError 则抛 McpError；若超时则抛带有 HTTP 408 码的 McpError；否则反序列化 result 部分并返回。
6. 清理：无论成功或失败，删除流和注册记录。

发送响应：_send_response()

当收到对端请求需回复时，根据成功或失败结果生成正确的 JSON-RPC 响应或错误。若是正常结果则构造 JSONRPCResponse（含 ID + result 字段），若是错误则构造 JSONRPCError，最后写入 _write_stream。

发送通知：send_notification()

对于只需单向推送的事件（如日志、进度、资源变化），该方法构造 JSONRPCNotification（无 ID），部分传输层可携带 related_request_id，直接写入 _write_stream 而不等待响应。

消息接收循环：_receive_loop()

此方法不断从底层流读取 JSON-RPC 消息，按类型分流到正确管道：

1. JSONRPCRequest：校验后包装成 RequestResponder 放入 _in_flight，调用 _received_request 钩子。若未完成响应，传给 _handle_incoming。
2. JSONRPCNotification：校验后调用 _received_notification 钩子，然后 _handle_incoming。若是取消通知，查找并取消对应的 in-flight 请求。
3. JSONRPCResponse / JSONRPCError：根据 ID 查找 _response_streams，将结果写入流供 send_request 读取。若找不到对应流，转交 _handle_incoming 处理。

可重写的钩子方法

BaseSession 定义了可重用钩子方法，将处理逻辑留给子类实现。ClientSession 和 ServerSession 重写这些方法注入初始化握手、工具调用分发等行为，无需触及底层 JSON-RPC 和并发细节。

四、ClientSession 和 ServerSession 的上层实现

BaseSession 做好了消息帧化、ID 跟踪、并发读写、超时与错误处理。上层只需继承并重写少数钩子即可快速实现会话逻辑。

客户端 (ClientSession)

初始化阶段通过构造函数注入下游读写流、超时设置及各类回调。调用 initialize() 时，向服务端发送 initialize 请求，收到 InitializeResult 后发送 InitializedNotification 完成握手。

在消息接收方面，ClientSession 重写 _received_request 钩子。当服务端发来 sampling/createMessage、roots/list 或 ping 等请求时，提取请求类型和参数，调用相应回调并将返回值传给 responder.respond() 完成响应。

对于通知，目前仅对日志消息作出处理，其他未处理的消息委托给构造时提供的 message_handler，以便上层监控或自定义处理。

服务端 (ServerSession)

握手逻辑内嵌在 _received_request 和 _received_notification 钩子里。首次碰到 InitializeRequest 时保存参数并返回 IncludeResult。接收到 InitializedNotification 时状态机进入 Initialized 阶段，此前到来的其他请求会触发 RuntimeError。

握手完成后，check_client_capability 方法读取客户端声明的能力集，判断功能可用性。ServerSession 提供了一系列高层接口，如 send_log_message、send_resource_updated 等，转换为对应的 ServerNotification 发送；create_message、list_roots 等方法封装了对 JSON-RPC 请求的构建与发送。

上行消息分发中，ServerSession 在 _handle_incoming 中将未处理的 RequestResponder 或 Notification 推送到内存队列，这是框架层面通过装饰器将请求自动路由到用户业务函数的基础。

从通用到可用的核心流程

典型部署中，Host 应用实例化 ClientSession 和 ServerSession，对接底层读写流。本地场景通常将标准输出连向标准输入，远程场景通过 HTTP 与 SSE 通道互联。

客户端调用 initialize() 发出请求，ServerSession 捕获并返回 InitializeResult，客户端再发送 InitializedNotification 完成握手。交互过程中，任何一方像调用本地函数一样发起高层 API 调用。例如客户端执行 await session.list_tools()，ClientSession 构造 JSON-RPC 请求发出，ServerSession 接收后封装成 RequestResponder 交给工具实现函数处理，执行完成后调用 responder.respond()，结果打包回传并在客户端被捕获反序列化。

同样地，服务端可直接调用 await server_session.create_message() 来触发大模型，支持上下文生成。上层只需注册回调，就能无缝插入新工具或扩展消息类型，BaseSession 的改动不影响业务逻辑。

client/session.py 和 server/session.py 在 BaseSession 底座上完成了握手、能力协商、消息分发、工具调用及高层接口封装，真正将 MCP 从协议文档变成了可扩展的 SDK。

总结

Session 类是 MCP SDK 在底层对 JSON-RPC 消息收发的统一封装，由框架内部创建和管理。编写 Server 代码时，只需关注声明工具函数、调用 mcp.run() 启动即可，无需看到 Session 的具体实现。只有在需要定制化底层消息拦截、自定义请求路由或替换 JSON-RPC 实现时，才需要考虑继承或修改 Session。