MCP Python SDK 协议层通过 BaseSession 类封装 JSON-RPC 消息收发、请求响应关联及通知分发逻辑。架构包含 Host、Client 和 Server，Client 负责与 Server 建立连接并交换消息。BaseSession 管理读写流、请求 ID 及异步任务，提供 send_request、send_notification 等核心方法。ClientSession 和 ServerSession 继承 BaseSession，分别处理握手、能力协商及工具调用等业务逻辑，实现开箱即用的客户端与服务端功能。
在 MCP 架构中，"Host"指的是最终承载大模型和用户界面的应用，比如桌面端的 Claude 客户端、IDE 插件或自研的聊天机器人后台。一个 Host 启动时，会在内部创建一个 MCP Client 实例，这个 Client 负责与外部的 MCP Server 建立一对一的连接（通常通过 stdio、HTTP+SSE 或其他支持的传输层）。
"接入层面的 Client"正是这个 MCP 客户端，它把上层业务或大模型交互中产生的"请求"（例如"列出可用工具""执行代码分析"）打包成符合 JSON-RPC 2.0 的消息，再通过底层传输通道发给 Server；同时，它也能接收来自 Server 的 Notification（如"工具状态更新"），并把所得数据交回给 Host，以便拼接到模型的 Prompt 中或做二次处理。
而 MCP"Server"则是真正提供上下文内容和工具执行能力的进程或服务。它在启动时会注册一系列接口（Resource 列表、工具调用、文件系统访问、外部 API 调用等），并监听来自 Client 的 JSON-RPC 消息。当收到"调用某个工具"这样的请求时，Server 会校验参数、调用相应逻辑，将结果封装成 Result 消息回传；如果出现异常，则返回带有标准错误码（如 ParseError、InvalidParams、MethodNotFound）的 Error 响应。
整个交互流程可以分为三个阶段。
在上面的 MCP 连接生命周期中，Client 和 Server 之间可以传递下列类型的 MCP 消息类型：
通过这种分层设计，MCP 将"业务逻辑"与"上下文 / 工具管理"解耦：Host 只管"我需要哪些信息 / 能力"；Server 负责"我怎样取到或实现这些能力"。最终，在 Host 内部，大模型得到的不再是死板的 API 返回值，而是经过精心组织的上下文提示，能够更灵活、更安全地调用外部工具、访问数据源。
(Protocol layer) 协议层负责消息的封装、请求 / 响应关联，以及高级通信模式的实现。其中的核心逻辑位于 BaseSession（定义在 mcp/shared/session.py）类，这里提供了所有 JSON-RPC 消息的读写、帧解析、请求–响应关联、通知分发的通用逻辑，是 MCP 协议层的通用实现，屏蔽了 JSON-RPC 消息的底层读写、请求–响应关联和通知分发细节。
BaseSession 类承担了对所有 JSON-RPC 消息的统一管理与生命周期控制。它通过一系列核心属性和方法来维护读写流、请求 ID、响应管道和后台任务栈，确保异步收发、请求–响应关联以及资源清理都井然有序。
它将"发送请求、等待响应""发出单向通知""接收并分发对端消息"这些通用逻辑全部封装，派生类（客户端或服务端的 Session）只需关注「收到请求／通知后如何处理」或「要发送哪些高层类型化消息」。
BaseSession 类的代码如下：
class Session(BaseSession[RequestT, NotificationT, ResultT]):
async def send_request(self, request: RequestT, result_type: type[Result]) -> Result:
"""发送请求并等待响应。若响应包含错误，则抛出 McpError。"""
# 请求发送和响应处理逻辑
pass
async def send_notification(self, notification: NotificationT) -> None:
"""发送无需响应的单向通知。"""
# 通知发送逻辑
pass
async def _received_request(self, responder: RequestResponder[ReceiveRequestT, ResultT]) -> None:
"""处理对方发来的请求。"""
# 收到请求后的业务处理
pass
async def _received_notification(self, notification: ReceiveNotificationT) -> None:
"""处理对方发来的通知。"""
# 收到通知后的业务处理
pass
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 类的核心属性包括：
_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 类中有一系列协议实现方法，主要包括 send_request()、send_notification()、_send_response() 和 _receive_loop() 等，用于控制 MCP 基本交互流程。
发送请求：send_request() 发送请求方法完成的是如何把一个高层的 Pydantic 请求模型打包成 JSON-RPC 消息、发出去，并在底层管理好"谁发的请求""等待谁的回应"这件事。
具体过程如下：
_request_id 自增取新 ID。_response_streams 中为该 ID 注册一个缓冲大小为 1 的内存流。_write_stream：发往对端。当收到对端的请求时，如何基于成功或失败结果，及时生成正确的 JSON-RPC 响应或错误，并写回去呢？
当收到对端发来的 Request，需要给它回复时：
_write_stream。在 send_notification() 方法中，会将那些只需单向推送、不需要回复的事件（如日志、进度、资源变化）等静悄悄地发给对端而不阻塞。具体实现上，是在方法内部构造 JSONRPCNotification（无 ID），部分传输层可携带 related_request_id（比如把进度通知关联到某次调用）；同时直接写入 _write_stream，不等待任何响应。
_receive_loop() 方法会不断从底层流里读入任意一方发来的 JSON-RPC 消息，按类型分流到"请求处理""通知处理"或"回应分发"的正确管道里。
该方法会持续读取 _read_stream 中的 SessionMessage，并根据消息类型分流。
_in_flight，先调用 _received_request(responder) 钩子，让子类有机会同步响应；如果钩子未完成响应，再把 responder 传给 _handle_incoming（供上层或框架线程处理）。_received_notification(notification) 钩子，然后 _handle_incoming。如果是取消通知，还会查找并取消对应的 in-flight 请求。_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 细节的认知成本。
BaseSession 将 JSON-RPC 的消息帧化、ID 跟踪、并发读写、超时与错误处理都做好了。上层只需继承并重写少数钩子，就能快速实现"MCP 客户端"或"MCP 服务端"的会话逻辑。
我们再来看 client/session.py 和 server/session.py 两个文件，是如何"填充"这一骨架，把 MCP 协议真正落地成可用的客户端与服务端逻辑的。
在 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 中，协议的握手逻辑被内嵌在 _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。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online