MCP 协议设计与实现

第11章 Python Client 实现剖析

作者杨艺韬 · 16,376 字 · 发布于 2026-04-16 · 更新于 2026-09-25

第 9 章我们分析了 TypeScript SDK 的 Client 实现，对 MCP 客户端的职责有了清晰认知。本章我们将目光转向 Python SDK 的客户端实现。

Python SDK 在设计哲学上与 TypeScript SDK 保持一致——都围绕"会话"这一核心抽象展开，但在具体实现上却因语言特性而呈现出显著差异。Python SDK 选择了 anyio 作为异步运行时抽象层，用 Pydantic 取代 Zod 进行类型校验，用 async with 上下文管理器替代手动的生命周期管理。这些选择不是偶然的，它们深刻影响了客户端的 API 形态和内部架构。

本章要点

ClientSession / ClientSessionGroup 两级抽象——单服务器会话 + 多服务器聚合
anyio + async with——结构化并发与资源生命周期的最佳实践
注册即声明——回调函数注册自动推断 Client capability
工具结果校验——基于 output_schema 的 JSON Schema 运行时验证（TS SDK 也有，实现路径不同，见 §11.6）
四种传输层：stdio / SSE / Streamable HTTP / WebSocket 的统一流抽象（基准版 3d7b311 里 websocket.py 尚未移除；2025-11-25 规范只定义 stdio 与 Streamable HTTP，SSE 是 2024-11-05 版的旧传输，WebSocket 从未进入规范）
OAuth2 认证：PKCE + TokenStorage + httpx 中间件
与 TS SDK 的 九个关键差异

11.1 客户端架构总览

Python MCP Client 的源码位于 src/mcp/client/ 目录下，核心文件包括：

session.py — ClientSession 类，单服务器会话的核心抽象
session_group.py — ClientSessionGroup 类，多服务器聚合管理
stdio.py — stdio 传输层实现
sse.py — SSE 传输层实现
streamable_http.py — Streamable HTTP 传输层实现
websocket.py — WebSocket 传输层实现（Python 独有）
src/mcp/client/auth/oauth2.py — OAuth2 客户端认证

下面这张架构图展示了各模块之间的关系：

graph TB
    subgraph 应用层
        APP[应用代码]
    end
    subgraph 会话层
        CSG[ClientSessionGroup<br/>多服务器聚合]
        CS[ClientSession<br/>单服务器会话]
    end
    subgraph 基础设施层
        BS[BaseSession<br/>请求/响应路由]
    end
    subgraph 传输层
        STDIO[stdio_client<br/>子进程通信]
        SSE[sse_client<br/>SSE 长连接]
        SH[streamable_http_client<br/>HTTP 流式]
        WS[websocket_client<br/>全双工]
    end
    subgraph 认证层
        OAUTH[OAuthClientProvider<br/>httpx.Auth · PKCE + 自动刷新]
    end
    APP --> CSG
    APP --> CS
    CSG -->|管理多个| CS
    CS -->|继承| BS
    CS --> STDIO
    CS --> SSE
    CS --> SH
    CS --> WS
    SH -.->|可选| OAUTH
    SSE -.->|可选| OAUTH

分层设计的三个核心选择

Python SDK 的分层反映了三个关键的设计哲学：

传输层完全抽象化——上层无需感知底层用的是 stdio 还是 HTTP，只看到 (read_stream, write_stream) 对
会话分级——单会话 (ClientSession) 管协议，多会话聚合 (ClientSessionGroup) 管应用层编排
资源管理结构化——从 AsyncExitStack 到 async with，Python 把"不泄漏资源"变成语言特性

11.2 ClientSession：单服务器会话

ClientSession（mcp-py-sdk/src/mcp/client/session.py:101）是 Python 客户端的核心类，它继承自 BaseSession（mcp-py-sdk/src/mcp/shared/session.py:166），负责与单个 MCP Server 的全部通信。我们先看其类型签名：

class ClientSession(
    BaseSession[
        types.ClientRequest,        # 发送的请求类型
        types.ClientNotification,   # 发送的通知类型
        types.ClientResult,         # 发送的结果类型
        types.ServerRequest,        # 接收的请求类型
        types.ServerNotification,   # 接收的通知类型
    ]
):

这个五元泛型参数看起来复杂，但逻辑清晰：前三个约束"我发出去的消息"，后两个约束"我收到的消息"。BaseSession 借这些泛型参数让类型检查器（pyright / mypy）在静态检查阶段约束消息类型；运行时的校验则由 _receive_request_adapter / _receive_notification_adapter 这两个 Pydantic TypeAdapter 完成。

11.2.1 构造与初始化

ClientSession 的构造函数（mcp-py-sdk/src/mcp/client/session.py:110）接受读写流和一系列回调函数：

def __init__(
    self,
    read_stream: ReadStream[SessionMessage | Exception],
    write_stream: WriteStream[SessionMessage],
    read_timeout_seconds: float | None = None,
    sampling_callback: SamplingFnT | None = None,
    elicitation_callback: ElicitationFnT | None = None,
    list_roots_callback: ListRootsFnT | None = None,
    logging_callback: LoggingFnT | None = None,
    message_handler: MessageHandlerFnT | None = None,
    client_info: types.Implementation | None = None,
    *,
    sampling_capabilities: types.SamplingCapability | None = None,
    experimental_task_handlers: ExperimentalTaskHandlers | None = None,
) -> None:

Protocol 定义回调类型

这些回调函数的设计值得关注。Python SDK 使用 Protocol 类来定义回调类型（SamplingFnT 见 mcp-py-sdk/src/mcp/client/session.py:24），这是 Python 式的结构化子类型（structural subtyping），思路与 TypeScript 按结构匹配的函数类型相同：

class SamplingFnT(Protocol):
    async def __call__(
        self,
        context: RequestContext[ClientSession],
        params: types.CreateMessageRequestParams,
    ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData: ...

Protocol 的好处：

无需继承，任何形状匹配的函数/类都符合协议
便于用普通函数作为回调
支持类型检查器做静态分析

默认回调的"优雅失败"

每个回调都有默认实现（_default_sampling_callback 见 mcp-py-sdk/src/mcp/client/session.py:63，构造函数里的 or 兜底见同文件 :127）。例如，当服务器发起 sampling 请求但客户端未注册回调时，默认返回错误：

async def _default_sampling_callback(
    context: RequestContext[ClientSession],
    params: types.CreateMessageRequestParams,
) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData:
    return types.ErrorData(
        code=types.INVALID_REQUEST,
        message="Sampling not supported",
    )

这种设计避免了未注册回调时的崩溃——明确返回"不支持"比抛出 NotImplementedError 更符合协议规范。

注册即声明：能力推断

初始化握手通过 initialize() 方法（mcp-py-sdk/src/mcp/client/session.py:148）完成。该方法会根据注册的回调函数自动推断客户端能力——判据就是"这个回调还是不是那个 _default_* 函数对象"：

async def initialize(self) -> types.InitializeResult:
    sampling = (
        (self._sampling_capabilities or types.SamplingCapability())
        if self._sampling_callback is not _default_sampling_callback
        else None
    )
    elicitation = (
        types.ElicitationCapability(form=types.FormElicitationCapability(), url=types.UrlElicitationCapability())
        if self._elicitation_callback is not _default_elicitation_callback
        else None
    )
    roots = (
        types.RootsCapability(list_changed=True)
        if self._list_roots_callback is not _default_list_roots_callback
        else None
    )
    # ... 组装 ClientCapabilities(sampling=..., elicitation=..., experimental=None,
    #     roots=..., tasks=self._task_handlers.build_capability()) 并发送 initialize 请求
    # （节选：省略了 roots 分支上的一段 TODO 注释与后半段的版本校验、initialized 通知）

如果注册了 sampling_callback，就声明支持 sampling 能力
如果注册了 list_roots_callback，就声明支持 roots 能力

这种"注册即声明"的设计避免了能力声明与实际实现不一致的问题。开发者只需要关心"我要不要支持这个功能"，不用单独维护 capability 声明。

以上握手按本专栏锚定的 2025-11-25 规范；2026-07-28 版已去掉 initialize / notifications/initialized 握手，版本与能力改由每个请求的 _meta 携带（SEP-2575）。

11.2.2 协议操作方法

ClientSession 为 MCP 协议的每种操作提供了对应的异步方法（以下方法依次定义在 mcp-py-sdk/src/mcp/client/session.py 的 :393、:299、:257、:278、:350、:358、:371、:219、:285、:292）：

方法	协议操作	返回类型
`list_tools()`	`tools/list`	`ListToolsResult`
`call_tool()`	`tools/call`	`CallToolResult`
`list_resources()`	`resources/list`	`ListResourcesResult`
`read_resource()`	`resources/read`	`ReadResourceResult`
`list_prompts()`	`prompts/list`	`ListPromptsResult`
`get_prompt()`	`prompts/get`	`GetPromptResult`
`complete()`	`completion/complete`	`CompleteResult`
`send_ping()`	`ping`	`EmptyResult`
`subscribe_resource()`	`resources/subscribe`	`EmptyResult`
`unsubscribe_resource()`	`resources/unsubscribe`	`EmptyResult`

以 call_tool 为例，它展示了客户端的结构化内容校验（TS SDK 同样有，差别见 §11.6）：

# src/mcp/client/session.py:299（省略 docstring）
async def call_tool(
    self,
    name: str,
    arguments: dict[str, Any] | None = None,
    read_timeout_seconds: float | None = None,
    progress_callback: ProgressFnT | None = None,
    *,
    meta: RequestParamsMeta | None = None,
) -> types.CallToolResult:
    result = await self.send_request(
        types.CallToolRequest(
            params=types.CallToolRequestParams(name=name, arguments=arguments, _meta=meta),
        ),
        types.CallToolResult,
        request_read_timeout_seconds=read_timeout_seconds,
        progress_callback=progress_callback,
    )
    if not result.is_error:
        await self._validate_tool_result(name, result)
    return result

call_tool 本身在 mcp-py-sdk/src/mcp/client/session.py:299；它调用的 _validate_tool_result（同文件 :324）会根据工具的 output_schema 对返回的 structured_content 进行 JSON Schema 校验。

Schema 缓存机制

工具的 output schema 在 list_tools() 调用时被缓存到 _tool_output_schemas 字典中（字段声明见 mcp-py-sdk/src/mcp/client/session.py:133，写入点在 list_tools 里，见同文件 :407），后续的 call_tool 会自动利用这个缓存。

async def _validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
    """Validate the structured content of a tool result against its output schema."""
    if name not in self._tool_output_schemas:
        # refresh output schema cache
        await self.list_tools()

    output_schema = None
    if name in self._tool_output_schemas:
        output_schema = self._tool_output_schemas.get(name)
    else:
        logger.warning(f"Tool {name} not listed by server, cannot validate any structured content")

    if output_schema is not None:
        from jsonschema import SchemaError, ValidationError, validate

        if result.structured_content is None:
            raise RuntimeError(f"Tool {name} has an output schema but did not return structured content")
        try:
            validate(result.structured_content, output_schema)
        except ValidationError as e:
            raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")
        except SchemaError as e:
            raise RuntimeError(f"Invalid schema for tool {name}: {e}")

如果缓存中没有该工具的 schema（比如工具是后来动态添加的），会自动触发一次 list_tools() 刷新缓存；刷新后仍然查不到，就只打一条 warning、跳过校验，而不是让调用失败。这种"lazy + cache"的组合既保证了正确性又避免了每次都查询。注意 jsonschema 是在函数体内部才 import 的——它是可选依赖，只有真的需要校验时才加载。

11.2.3 服务器请求处理

MCP 是双向协议，服务器也会向客户端发起请求。_received_request 方法（mcp-py-sdk/src/mcp/client/session.py:415）通过 Python 的 match/case 模式匹配来分发这些请求：

# src/mcp/client/session.py:415（简化：省略类型注解与注释，Elicit / ListRoots 两个分支与 Sampling 同构）
async def _received_request(self, responder):
    ctx = RequestContext[ClientSession](request_id=responder.request_id, meta=responder.request_meta, session=self)

    # tasks/get、tasks/result、tasks/list、tasks/cancel 先被实验性的 task handler 截走
    if self._task_handlers.handles_request(responder.request):
        with responder:
            await self._task_handlers.handle_request(ctx, responder)
        return None

    match responder.request:
        case types.CreateMessageRequest(params=params):
            with responder:
                if params.task is not None:        # 任务增强的 sampling
                    response = await self._task_handlers.augmented_sampling(ctx, params, params.task)
                else:
                    response = await self._sampling_callback(ctx, params)
                client_response = ClientResponse.validate_python(response)
                await responder.respond(client_response)
        case types.ElicitRequest(params=params):
            ...                                    # 同上，走 _elicitation_callback
        case types.ListRootsRequest():
            ...                                    # 同上，走 _list_roots_callback(ctx)
        case types.PingRequest():
            with responder:
                return await responder.respond(types.EmptyResult())
        case _:
            pass  # Task requests handled above by _task_handlers

注意 with responder 上下文管理器的使用。RequestResponder（mcp-py-sdk/src/mcp/shared/session.py:60）实现了 __enter__/__exit__（同文件 :99 与 :106），在进入时新建一个 anyio.CancelScope 作为取消作用域（同文件 :102），在退出时若请求已经应答，就通过 on_complete 把它从会话的 _in_flight 表里移除。回调抛异常时不会走到这一步——异常一路冒到 BaseSession._receive_loop，由那里的兜底分支回一条 -32602 "Invalid request parameters"（mcp-py-sdk/src/mcp/shared/session.py:377-388，实测 sampling 回调抛 ValueError 时 server 收到的就是这条）。

这一节讲的反向请求（sampling、elicitation、roots）按本专栏锚定的 2025-11-25 规范是 server 主动发给 client 的 JSON-RPC 请求；2026-07-28 版改成了 Multi Round-Trip Requests——server 在结果里返回 InputRequiredResult，client 带着 inputResponses 重试原请求（SEP-2322）；同一版还把 Roots、Sampling 与 Logging 三项特性标为 Deprecated（SEP-2577，弃用窗口内仍可用）。

match/case vs isinstance 链

Python 3.10+ 的 match/case 类模式不只是 isinstance 检查，还能用关键字模式顺手解构字段（按属性名取值，对任何类都适用，并非 Pydantic 特有）：

# match/case 风格：类型匹配 + 字段解构一气呵成
case types.CreateMessageRequest(params=params):
    response = await self._sampling_callback(ctx, params)

# isinstance 风格：冗余
if isinstance(req, types.CreateMessageRequest):
    params = req.params
    response = await self._sampling_callback(ctx, params)

11.3 ClientSessionGroup：多服务器聚合

真实的 AI Agent 应用往往需要同时连接多个 MCP Server。ClientSessionGroup 正是为此设计的——它管理多个 ClientSession，并将所有服务器的 tools、resources、prompts 聚合到统一的命名空间中。

graph LR
    subgraph ClientSessionGroup
        TOOLS[聚合 Tools 字典]
        RES[聚合 Resources 字典]
        PROMPTS[聚合 Prompts 字典]
    end
    subgraph ServerA[Server A - 文件系统]
        SA[ClientSession A]
        TA[read_file, write_file]
    end
    subgraph ServerB[Server B - 数据库]
        SB[ClientSession B]
        TB[query, execute]
    end
    subgraph ServerC[Server C - Web 搜索]
        SC[ClientSession C]
        TC[search, fetch]
    end
    SA --> TA
    SB --> TB
    SC --> TC
    TOOLS -->|路由| SA
    TOOLS -->|路由| SB
    TOOLS -->|路由| SC

11.3.1 连接管理与生命周期

ClientSessionGroup 实现了 async with 协议，用 AsyncExitStack 管理所有子会话的生命周期：

# 简化自 src/mcp/client/session_group.py:85-148：源码把字段类型注解写在类体里、
# __init__ 里只做赋值，这里合并写在一起；_session_exit_stacks 与 _component_name_hook 省略
class ClientSessionGroup:
    def __init__(
        self,
        exit_stack: contextlib.AsyncExitStack | None = None,
        component_name_hook: _ComponentNameHook | None = None,
    ):
        self._tools: dict[str, types.Tool] = {}
        self._resources: dict[str, types.Resource] = {}
        self._prompts: dict[str, types.Prompt] = {}
        self._sessions: dict[ClientSession, _ComponentNames] = {}
        self._tool_to_session: dict[str, ClientSession] = {}

        if exit_stack is None:
            self._exit_stack = contextlib.AsyncExitStack()
            self._owns_exit_stack = True
        else:
            self._exit_stack = exit_stack
            self._owns_exit_stack = False

这里有一个精妙的设计：exit_stack 可以外部传入，也可以内部创建。

场景	`exit_stack`	`owns`	生命周期
独立使用	内部创建	True	`async with` 结束时关闭
嵌入到更大的资源管理	外部传入	False	由调用者控制

这使得 ClientSessionGroup 可以嵌入到更大的资源管理体系中——比如一个 Agent 框架可能有自己的全局 exit_stack，所有资源都统一管理。

两级 exit_stack 架构

仔细看 ClientSessionGroup 的字段声明（src/mcp/client/session_group.py:119-120），实际上维护着两级 exit_stack：

_exit_stack: contextlib.AsyncExitStack                               # 主 stack
_session_exit_stacks: dict[mcp.ClientSession, contextlib.AsyncExitStack]  # 每会话 stack

两级设计的理由：

主 _exit_stack——管理 Group 整体的 entering/exiting。如果 Group owns 它、__aexit__ 时 aclose()；不 own 就让外部处理。
_session_exit_stacks[session]——每个 ClientSession 有自己独立的 exit stack，记录它的底层连接资源（stdio 进程、HTTP 连接、SSE 流）。

这让"断开一个 session 但保留其他 session"成为可能——disconnect_from_server(session) 只关这一个 session 的 exit_stack、其他 session 继续运行。如果只有一个共享 stack、就得全部关或都不关。

看上去并发，默认用法下其实串行

看 __aexit__ 的真实实现（src/mcp/client/session_group.py:156-171）：

# 省略了签名的类型注解与 docstring，其余逐行照录
async def __aexit__(self, _exc_type, _exc_val, _exc_tb) -> bool | None:
    # Only close the main exit stack if we created it
    if self._owns_exit_stack:
        await self._exit_stack.aclose()

    # Concurrently close session stacks.
    async with anyio.create_task_group() as tg:
        for exit_stack in self._session_exit_stacks.values():
            tg.start_soon(exit_stack.aclose)

单看后半段，anyio.create_task_group() + tg.start_soon(exit_stack.aclose) 是想让所有 session 的关闭并发发生。但要连着 _establish_session 一起读（src/mcp/client/session_group.py:320-323）：每个 session_stack 建好后都被 self._exit_stack.enter_async_context(session_stack) 注册进了主 stack。于是：

默认用法（Group 自己创建 exit_stack）：第一步 self._exit_stack.aclose() 就已经按后进先出把各个 session_stack 逐个串行关掉了，后面 task group 里的 aclose() 面对的是已关闭的 stack，只是空操作。实测连 3 个 stdio server（各自在 stdin 关闭后还要 1.5 秒才退出），退出 async with ClientSessionGroup() 用了 4.75 秒——正是三段等待相加。关闭一个 stdio 会话最多要先等 PROCESS_TERMINATION_TIMEOUT = 2.0 秒，超时后 _terminate_process_tree（src/mcp/client/stdio.py:255）还要再给 2 秒，连的 server 越多，退出越慢。
外部传入 exit_stack：主 stack 不在这里关，才真的走并发分支。但实测对 stdio 连接会抛 RuntimeError: Attempted to exit cancel scope in a different task than it was entered in——stdio_client 里的 anyio task group 必须在进入它的那个任务里退出，放进 task group 的子任务里 aclose() 违反了这条规则。

所以这段"并发关闭"在基准版里并没有兑现预期：默认路径是串行的，另一条路径对 stdio 会报错。

`disconnect_from_server` 的 4 态检查

src/mcp/client/session_group.py:213-223 的 disconnect_from_server 实现：

async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
    session_known_for_components = session in self._sessions
    session_known_for_stack = session in self._session_exit_stacks

    if not session_known_for_components and not session_known_for_stack:
        raise MCPError(
            code=types.INVALID_PARAMS,
            message="Provided session is not managed or already disconnected.",
        )
    # ... 分别处理两个维度

两个独立维度：_sessions 记录"组件归属"、_session_exit_stacks 记录"资源所有权"。两者正常情况下同步更新、但边缘情况可能脱同步（比如 connect 过程中异常、一个 dict 已插入另一个还没）。

代码逻辑明确说"只要还在任一维度里就不是完全未知"——只有两者都没时才抛 INVALID_PARAMS。这让 disconnect 的幂等性更强——用户可以安全地重复调 disconnect_from_server(same_session)，第二次会抛合适的错误而不是静默把已清理的字典再搅一遍导致 KeyError。

11.3.2 命名冲突：`component_name_hook` 的作用

同一个工具名（比如 search）被两个不同服务器暴露时会怎样？答案是 _aggregate_components 会直接抛 MCPError(code=INVALID_PARAMS)——它在合并前先算 tools_temp.keys() & self._tools.keys()，交集非空就报 "… already exist in group tools."，prompts / resources 同理。也就是说 SDK 选择了「宁可连接失败也不静默覆盖」。

component_name_hook 参数（src/mcp/client/session_group.py:122-125）给用户一个主动命名空间化的钩子：

# Optional fn consuming (component_name, server_info) for custom names.
# This is to provide a means to mitigate naming conflicts across servers.
# Example: (tool_name, server_info) => "{result.server_info.name}.{tool_name}"
_ComponentNameHook: TypeAlias = Callable[[str, types.Implementation], str]

源码注释直接给了典型用法：

name_fn = lambda name, server_info: f"{server_info.name}_{name}"
async with ClientSessionGroup(component_name_hook=name_fn) as group:
    for params in server_params:
        await group.connect_to_server(params)

这样 Server A 的 search 注册为 "ServerA_search"、Server B 的 search 注册为 "ServerB_search"——两者共存。

注意 call_tool 路由是按"外部名"走的（src/mcp/client/session_group.py:203）：

# 简化自 src/mcp/client/session_group.py:193-211
async def call_tool(self, name, arguments=...):
    session = self._tool_to_session[name]          # 按外部名查 session
    session_tool_name = self.tools[name].name       # 取出 session 里的原名
    return await session.call_tool(session_tool_name, ...)

_tool_to_session[name] 用外部（可能已重命名的）名做 lookup、self.tools[name].name 取出 session 内部的原名——因为 session 只认自己当初注册的 search、不认 ServerA_search。这种"外部统一命名 + 内部原名转发"的设计让 rename hook 只影响 Group 层、不扰动底层 session 协议。调试时注意两层之间的映射关系。

使用示例

连接新服务器通过 connect_to_server 完成：

async with ClientSessionGroup() as group:
    # 连接第一个服务器
    session_a = await group.connect_to_server(
        StdioServerParameters(command="python", args=["-m", "my_server"])
    )

    # 连接第二个服务器
    session_b = await group.connect_to_server(
        StreamableHttpParameters(url="http://localhost:8080/mcp")
    )

    # 连接第三个服务器
    session_c = await group.connect_to_server(
        SseServerParameters(url="http://legacy.example.com/sse")
    )

    # 此时 group.tools 包含三个服务器的所有工具
    print(f"Available tools: {list(group.tools.keys())}")

    # 调用工具（自动路由到正确的服务器）
    result = await group.call_tool("read_file", {"path": "/tmp/test.txt"})

传输层自动选择

_establish_session 方法展示了传输层的选择逻辑：

# src/mcp/client/session_group.py:264
async def _establish_session(
    self,
    server_params: ServerParameters,
    session_params: ClientSessionParameters,
) -> tuple[types.Implementation, mcp.ClientSession]:
    """Establish a client session to an MCP server."""
    session_stack = contextlib.AsyncExitStack()
    try:
        if isinstance(server_params, StdioServerParameters):
            client = mcp.stdio_client(server_params)
            read, write = await session_stack.enter_async_context(client)
        elif isinstance(server_params, SseServerParameters):
            client = sse_client(url=server_params.url, headers=server_params.headers, ...)
            read, write = await session_stack.enter_async_context(client)
        else:                                   # StreamableHttpParameters
            httpx_client = create_mcp_http_client(...)
            await session_stack.enter_async_context(httpx_client)
            client = streamable_http_client(
                url=server_params.url,
                http_client=httpx_client,
                terminate_on_close=server_params.terminate_on_close,
            )
            read, write = await session_stack.enter_async_context(client)

        session = await session_stack.enter_async_context(mcp.ClientSession(read, write, ...))
        result = await session.initialize()

        self._session_exit_stacks[session] = session_stack
        await self._exit_stack.enter_async_context(session_stack)
        return result.server_info, session
    except Exception:
        await session_stack.aclose()
        raise

三点值得注意。第一，分发用的是 isinstance 链而不是 match/case，且 Streamable HTTP 走的是 else 兜底分支——函数名是 streamable_http_client（不是 streamablehttp_client）。第二，每个会话都有独立的 AsyncExitStack，确保单个服务器断连不会影响其他会话；这个 stack 建好后还会被注册进 Group 的主 _exit_stack。第三，_establish_session 返回的是 (server_info, session)——server_info 来自 initialize() 的结果，正是 component_name_hook 拿到的第二个参数。

常见的命名策略

回到 component_name_hook——除了前面示例里的服务器名前缀，还可以选：

{server}_{tool} — 服务器名前缀，最直观
{tool}@{server} — 装饰符风格
{tool} + 版本号后缀 — 保持名称简洁
自定义映射表 — 精细控制

原子性保证

聚合过程使用临时字典来保证原子性——新 server 的组件与已有组件重名时直接报错，已有的聚合状态不会被污染：

# 简化自 src/mcp/client/session_group.py:332-405（resources / tools 两段与 prompts 同构）
async def _aggregate_components(self, server_info, session):
    component_names = self._ComponentNames()      # 反向索引：这个 session 注册了哪些名字

    # 临时收集，不直接动聚合字典
    prompts_temp: dict[str, types.Prompt] = {}
    resources_temp: dict[str, types.Resource] = {}
    tools_temp: dict[str, types.Tool] = {}
    tool_to_session_temp: dict[str, mcp.ClientSession] = {}

    # 依次拉 prompts → resources → tools；某一类拉取失败只记 warning、跳过
    try:
        for prompt in (await session.list_prompts()).prompts:
            name = self._component_name(prompt.name, server_info)
            prompts_temp[name] = prompt
            component_names.prompts.add(name)
    except MCPError as err:
        logging.warning(f"Could not fetch prompts: {err}")
    ...  # resources、tools 同理；tools 还会写 tool_to_session_temp

    # 一样都没拉到，就把这个 session 的 exit stack 从登记表里去掉
    if not any((prompts_temp, resources_temp, tools_temp)):
        del self._session_exit_stacks[session]

    # 检查冲突（在合并前）：prompts、resources、tools 各查一次
    matching_tools = tools_temp.keys() & self._tools.keys()
    if matching_tools:
        raise MCPError(code=types.INVALID_PARAMS, message=f"{matching_tools} already exist in group tools.")

    # 全部检查通过才提交
    self._sessions[session] = component_names
    self._prompts.update(prompts_temp)
    self._resources.update(resources_temp)
    self._tools.update(tools_temp)
    self._tool_to_session.update(tool_to_session_temp)

"原子"指的是冲突检测这一层：检测不过就抛错，聚合字典一个都没改。拉取阶段的 MCPError 则被吞掉只记 warning——某个 server 的 prompts/list 失败，它的 tools 照样会被合并进来。

11.3.3 动态断连

disconnect_from_server 支持运行时动态移除某个服务器。它通过 _ComponentNames 反向索引快速定位该会话注册的所有组件，逐一清理：

# 简化自 ClientSessionGroup.disconnect_from_server，src/mcp/client/session_group.py:213-246：
# 省略了 §11.3.1 那段"两个维度都不认识就抛错"的检查，以及每次 del 前的 `if name in ...` 保护；
# 源码删除顺序是 prompts → resources → tools
async def disconnect_from_server(self, session: ClientSession) -> None:
    # 取出该会话注册的所有组件名
    component_names = self._sessions.pop(session)

    # 从聚合字典中删除
    for name in component_names.tools:
        del self._tools[name]
        del self._tool_to_session[name]
    for name in component_names.resources:
        del self._resources[name]
    for name in component_names.prompts:
        del self._prompts[name]

    # 关闭该会话的 exit_stack（不影响其他会话）
    session_stack = self._session_exit_stacks.pop(session)
    await session_stack.aclose()

这个设计支持热插拔——Agent 运行过程中可以动态加入或移除 MCP Server，无需重启。

11.4 传输层：四种连接方式

Python SDK 提供了四种传输层实现，它们都遵循相同的模式——作为异步上下文管理器，yield 出 (read_stream, write_stream) 元组。

11.4.1 stdio 传输

stdio_client 通过 _create_platform_compatible_process（Unix 下开新 session/进程组，Windows 下放进 Job Object）启动子进程，再把子进程的 stdin/stdout 包装为 MCP 消息流：

# src/mcp/client/stdio.py:104（节选：省略类型注解、进程创建失败时的流清理，
# 以及两个协程外层的 async with / ClosedResourceError 处理）
@asynccontextmanager
async def stdio_client(server: StdioServerParameters, errlog: TextIO = sys.stderr):
    read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
    write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

    command = _get_executable_command(server.command)
    process = await _create_platform_compatible_process(
        command=command, args=server.args, env=..., errlog=errlog, cwd=server.cwd,
    )

    async def stdout_reader():
        buffer = ""
        async for chunk in TextReceiveStream(process.stdout, encoding=server.encoding, ...):
            lines = (buffer + chunk).split("\n")
            buffer = lines.pop()
            for line in lines:
                try:
                    message = types.jsonrpc_message_adapter.validate_json(line, by_name=False)
                except Exception as exc:
                    await read_stream_writer.send(exc)   # 解析失败：把异常对象送进读流（见 §11.10.3）
                    continue
                await read_stream_writer.send(SessionMessage(message))

    async def stdin_writer():
        async for session_message in write_stream_reader:
            json = session_message.message.model_dump_json(by_alias=True, exclude_unset=True)
            await process.stdin.send((json + "\n").encode(encoding=server.encoding, ...))

    async with anyio.create_task_group() as tg, process:
        tg.start_soon(stdout_reader)
        tg.start_soon(stdin_writer)
        try:
            yield read_stream, write_stream
        finally:
            ...  # 见下面的关闭序列

注意 stdout_reader 里的 buffer = lines.pop()：split 之后最后一段是尚未收到换行的残片，必须留到下一个 chunk 再拼——这是行分帧协议里最常见的一个坑。

Graceful Shutdown

finally 块里遵循 MCP 规范的 stdio 关闭序列（src/mcp/client/stdio.py:185-211）：

finally:
    # MCP spec: stdio shutdown sequence
    # 1. Close input stream to server
    # 2. Wait for server to exit, or send SIGTERM if it doesn't exit in time
    # 3. Send SIGKILL if still not exited
    if process.stdin:
        try:
            await process.stdin.aclose()
        except Exception:
            pass

    try:
        with anyio.fail_after(PROCESS_TERMINATION_TIMEOUT):   # = 2.0 秒
            await process.wait()
    except TimeoutError:
        await _terminate_process_tree(process)                # SIGTERM → 等待 → SIGKILL
    except ProcessLookupError:
        pass
    await read_stream.aclose()
    ...

这种"三段式关闭"是 Unix 进程管理的标准实践——先温和，再严厉，最后暴力。值得注意的是第三步不是 process.kill() 而是 _terminate_process_tree()：POSIX 下走 os.killpg(pgid, SIGTERM) 对整个进程组下手，进程组 2 秒内还没退净就 SIGKILL（src/mcp/os/posix/utilities.py:13-60）。这样做是因为 MCP server 常常是 npx / uvx 这类包装脚本，只杀掉直接子进程会留下孤儿。

11.4.2 SSE 与 Streamable HTTP 传输

SSE 传输使用 httpx + httpx-sse 库建立长连接。Streamable HTTP 传输是 SSE 的演进版本，支持双向流式通信，并用 mcp-session-id 头来维护会话状态（这是本专栏锚定的 2025-11-25 规范；2026-07-28 版已去掉协议级会话与 Mcp-Session-Id，SEP-2567）。两者都可以通过 OAuthClientProvider 这个 httpx.Auth 中间件进行认证：sse_client 直接收 auth 参数，streamable_http_client 则要把它配在传入的 httpx.AsyncClient 上。

11.4.3 WebSocket 传输

基准版 Python SDK 独有的 WebSocket 实现（上游已于 2026-06-08 的 PR #2785 移除，详见第 14 章），提供真正的全双工通信。

11.4.4 传输参数模型

src/mcp/client/session_group.py:67 把传输参数统一成一个 ServerParameters 类型别名——注意只有三种，WebSocket 不在其中：

ServerParameters: TypeAlias = StdioServerParameters | SseServerParameters | StreamableHttpParameters

这使得 connect_to_server 可以接受任意一种参数，内部通过 isinstance 分发到对应的传输层。上一小节的 websocket_client 虽然还在 src/mcp/client/websocket.py 里，却没有对应的参数模型——全文件 grep 不到 websocket 字样，所以 ClientSessionGroup 管不了 WebSocket 连接；要用它只能自己手写 async with websocket_client(url) as (read, write) 再建 ClientSession。这也是"WebSocket 不是规范定义的传输"在 SDK 里的一处具体体现。

11.5 OAuth2 客户端认证

Python SDK 的 OAuth2 实现位于 src/mcp/client/auth/oauth2.py，支持 Authorization Code + PKCE 流程。

PKCE 参数生成

PKCEParameters 类封装了 code_verifier/code_challenge 的生成：

# src/mcp/client/auth/oauth2.py:56
class PKCEParameters(BaseModel):
    code_verifier: str = Field(..., min_length=43, max_length=128)
    code_challenge: str = Field(..., min_length=43, max_length=128)

    @classmethod
    def generate(cls) -> "PKCEParameters":
        code_verifier = "".join(
            secrets.choice(string.ascii_letters + string.digits + "-._~")
            for _ in range(128)
        )
        digest = hashlib.sha256(code_verifier.encode()).digest()
        code_challenge = base64.urlsafe_b64encode(digest).decode().rstrip("=")
        return cls(code_verifier=code_verifier, code_challenge=code_challenge)

TokenStorage 接口

认证流程通过 TokenStorage 协议实现 token 的持久化存储：

class TokenStorage(Protocol):
    async def get_tokens(self) -> OAuthToken | None: ...
    async def set_tokens(self, tokens: OAuthToken) -> None: ...
    async def get_client_info(self) -> OAuthClientInformationFull | None: ...
    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None: ...

应用可以自定义存储后端：

文件（开发环境）
系统密钥链（Keychain/Credentials Manager）
数据库（多用户场景）
内存（测试）

httpx 中间件集成

Python SDK 把整个 OAuth 客户端做成了一个 httpx 认证中间件—— src/mcp/client/auth/oauth2.py 里的 OAuthClientProvider(httpx.Auth)。入口是 async_auth_flow，一个异步生成器：yield 出去一个请求，httpx 替它发出去，再把响应送回来。整个 OAuth 状态机就写在这一个生成器里。

# 简化自 src/mcp/client/auth/oauth2.py:506 起的 async_auth_flow（注释为本专栏所加）
class OAuthClientProvider(httpx.Auth):
    async def async_auth_flow(self, request):
        async with self.context.lock:                 # 全流程串行，避免并发重复换 token
            if not self._initialized:
                await self._initialize()

            self.context.protocol_version = request.headers.get(MCP_PROTOCOL_VERSION)

            # 已知过期且能刷 —— 先刷，不等 401
            if not self.context.is_token_valid() and self.context.can_refresh_token():
                refresh_response = yield await self._refresh_token()
                if not await self._handle_refresh_response(refresh_response):
                    self._initialized = False          # 刷失败，退回完整授权

            if self.context.is_token_valid():
                self._add_auth_header(request)

            response = yield request

            if response.status_code == 401:
                ...                                    # 完整 OAuth 流程，见下
                self._add_auth_header(request)
                yield request                          # 带新 token 重试一次

有三个细节值得注意：

1. 刷新是前置的，不是等 401 才做。 token 已知过期且 can_refresh_token() 为真时，中间件先发刷新请求，再发业务请求。401 分支处理的是另一类情况——服务端不认这个 token，那已经不是刷新能解决的了。

2. 401 走的是完整重新授权，不是"再刷一次"。 源码里这段有五个带编号的步骤：

步骤	做的事
1	从 `WWW-Authenticate` 取 resource metadata URL，按 SEP-985 的候选列表逐个探测 PRM；拿到后按 RFC 8707 校验 resource 与 server URL 匹配
2	用 PRM 里的 `authorization_servers[0]` 去发现 OASM，同样带一组回退 URL 兼容老服务端
3	综合 `WWW-Authenticate` 的 scope、PRM、OASM 和本地 `grant_types`，决定最终请求哪些 scope
4	还没有 client_info 就注册：优先用 URL 形式的 client ID（CIMD），不支持则回退到动态客户端注册（DCR）
5	执行授权码流程并换取 token

走完之后才 _add_auth_header 重试原请求——只重试一次，再失败就把响应交回上层。

3. 全程持有 self.context.lock。 多个并发请求同时撞上过期 token 时，只有一个会真正去换，其余在锁上等——避免把授权服务器打成 N 倍请求，也避免几个协程各自写 TokenStorage 互相覆盖。

顺带一提：403 有独立分支，会从 WWW-Authenticate 里取 error 字段单独处理—— 权限不足和未认证是两回事，不该用同一套重试逻辑对待。

11.6 与 TypeScript Client 的核心差异

通过对比两个 SDK 的客户端实现，可以提炼出以下关键差异：

维度	Python SDK	TypeScript SDK
异步运行时	anyio（兼容 asyncio/trio）	原生 async/await
生命周期管理	`async with` + `AsyncExitStack`	手动 `connect()`（`mcp-ts-sdk/packages/client/src/client/client.ts:492`）/ `close()`（`mcp-ts-sdk/packages/core/src/shared/protocol.ts:734`）
类型校验	Pydantic `TypeAdapter`	Zod schema
回调类型	`Protocol`（结构化子类型）	TypeScript 函数类型
消息分发	`match/case` 模式匹配	按 method 查 `_requestHandlers` 表（`mcp-ts-sdk/packages/core/src/shared/protocol.ts:543`）
多服务器	内置 `ClientSessionGroup`（`mcp-py-sdk/src/mcp/client/session_group.py`）	需自行实现
工具结果校验	内置 `_validate_tool_result`，调用时按需 `import jsonschema`（`mcp-py-sdk/src/mcp/client/session.py:324`）	同样内置：`callTool` 走预编译的 `JsonSchemaValidator`（`mcp-ts-sdk/packages/client/src/client/client.ts:882`）
传输层切换	`isinstance` + 上下文管理器	Transport 接口
WebSocket	✅ 支持（`mcp-py-sdk/src/mcp/client/websocket.py`）	❌ 不支持（`mcp-ts-sdk/packages/client/src/client/` 下只有 stdio / sse / streamableHttp 三种传输）

最关键的三点差异

差异一：生命周期管理

Python 的 async with 将资源获取与释放绑定在语法结构上，不可能忘记关闭连接：

async with ClientSession(read, write) as session:
    await session.initialize()
    result = await session.call_tool(...)
# 退出 block 时自动清理，无论是否抛异常

TypeScript 需要开发者自觉调用 close()，或使用 try/finally：

const client = new Client(...);
try {
    await client.connect(transport);
    const result = await client.callTool(...);
} finally {
    await client.close();  // 容易忘
}

AsyncExitStack 的嵌套使用更是 Python 独有的模式——ClientSessionGroup 用主 exit_stack 管理所有子会话的 exit_stack，形成资源管理的树状结构。

差异二：多服务器聚合

ClientSessionGroup 是 Python SDK 独有的抽象。它不仅管理连接，还提供了：

组件聚合（tools/resources/prompts 统一命名空间）
命名冲突检测
动态断连
原子性保证

TypeScript SDK 没有对应的内置实现，开发者需要自行管理多个 Client 实例。这也是 MCP 框架层的机会——多 server 管理往往由 Agent 框架在自己那一层实现。

差异三：工具结果校验的实现路径（而不是有无）

这一条容易想当然地写成"Python 有、TypeScript 没有"——两边其实都校验，差别在于校验器什么时候被准备好。

Python 的 call_tool 在返回结果前调 _validate_tool_result（mcp-py-sdk/src/mcp/client/session.py:324），拿 _tool_output_schemas 缓存里的 schema 现场喂给 jsonschema.validate；jsonschema 是函数体内才 import 的可选依赖。缓存里没有该工具时会先补一次 list_tools()，仍然查不到就只打 warning 放行。

# Python：schema 缓存 + 现场 jsonschema.validate
result = await session.call_tool("calc", {"a": 1, "b": 2})
# 如果 result.structured_content 缺失或不符合 schema，抛 RuntimeError

TypeScript 侧是预编译路线：listTools() 返回后立刻调 cacheToolMetadata（mcp-ts-sdk/packages/client/src/client/client.ts:999），把每个带 outputSchema 的工具编译成一个 JsonSchemaValidator（默认实现是 AjvJsonSchemaValidator）存进 _cachedToolOutputValidators；callTool 里取出验证器直接跑（mcp-ts-sdk/packages/client/src/client/client.ts:882），校验失败抛 ProtocolError。"工具声明了 outputSchema 却没返回 structured content（且不是 error）就报错"这条硬约束两边都有：TS 抛 ProtocolError，Python 抛 RuntimeError（mcp-py-sdk/src/mcp/client/session.py:339）。

// TypeScript：listTools 时预编译 validator，callTool 时直接跑
const result = await client.callTool({ name: "calc", arguments: { a: 1, b: 2 } });
// structuredContent 不符合 schema 时抛 ProtocolError

所以真正的差异是"惰性 import + 每次现场校验"对"listTools 时预编译 + 调用时零额外解析"——后者省掉了重复编译 schema 的开销，前者省掉了没用到校验时的依赖。

11.7 实战：构建一个多 MCP Server 的 Agent

下面展示一个真实场景——Agent 同时连接文件系统、数据库、Web 搜索三个 MCP Server：

import anyio
from mcp import ClientSessionGroup
from mcp.client.session_group import (
    StdioServerParameters,
    StreamableHttpParameters,
)

# 示意代码：API_KEY 与 llm_decide_tool（让 LLM 选工具）需自行提供
async def run_agent():
    # 命名策略：用服务器名前缀避免冲突
    name_fn = lambda name, info: f"{info.name}_{name}"

    async with ClientSessionGroup(component_name_hook=name_fn) as group:
        # 文件系统 MCP Server（本地 stdio）
        fs_session = await group.connect_to_server(
            StdioServerParameters(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
            )
        )

        # 数据库 MCP Server（本地 stdio）
        db_session = await group.connect_to_server(
            StdioServerParameters(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-postgres", "postgres://..."],
            )
        )

        # Web 搜索 MCP Server（远程 HTTP）
        web_session = await group.connect_to_server(
            StreamableHttpParameters(
                url="https://search.example.com/mcp",
                headers={"Authorization": f"Bearer {API_KEY}"},
            )
        )

        print(f"Connected. Tools: {list(group.tools.keys())}")
        # 每个工具名都带上 "<serverInfo.name>_" 前缀；前缀取的是各 server 在
        # initialize 响应里自报的 serverInfo.name，不是这里的变量名

        # 主 Agent 循环
        while True:
            user_input = input("You: ")
            if user_input == "exit":
                break

            # 让 LLM 决定调用哪个工具
            decision = await llm_decide_tool(user_input, group.tools)

            if decision.tool_name:
                result = await group.call_tool(
                    decision.tool_name,
                    decision.arguments,
                )
                print(f"Tool result: {result.content}")
            else:
                print(f"Bot: {decision.text}")

anyio.run(run_agent)

这个模式的威力：

统一调用接口——不管工具在哪个 Server，都是 group.call_tool()
动态扩展——可以运行时加入新 Server
故障隔离——单个 Server 挂掉不影响其他

11.8 四个反模式

11.8.1 反模式一：忘记 async with

# ❌ 不进入上下文：BaseSession.__aenter__ 才会启动 _receive_loop，
#    没有它就没人读响应，initialize() 会一直挂着（实测）
session = ClientSession(read, write)
await session.initialize()

# ✅ 使用 async with
async with ClientSession(read, write) as session:
    await session.initialize()
    # 自动关闭

11.8.2 反模式二：在反向请求回调里等同一个 session 的响应

先澄清一个常见误解：并发调用同一个 session 不需要加锁。send_request 在第一个 await 之前就同步分配好自增的请求 id，并为这个 id 建一个容量为 1 的专属响应流（src/mcp/shared/session.py:254-258），响应回来按 id 投递。实测 30 个并发 call_tool、server 故意乱序回包，结果零错配。

真正的坑在另一个方向。BaseSession._receive_loop 收到 server 发来的请求后，是直接 await self._received_request(responder)（src/mcp/shared/session.py:373），不另起任务——回调跑多久，接收循环就停多久。于是：

回调执行期间，client 自己发出去的请求即使 server 已经回了，响应也要等回调结束才被读到（实测：sampling 回调睡 2 秒，期间发出的 send_ping() 也拖到 2 秒后才返回）；
回调里再对同一个 session 发请求并等待结果（比如在 sampling_callback 里调 ctx.session.send_ping() 或 call_tool()），会直接死锁：响应永远没人读，而默认又不设超时（实测一直挂到外层取消）。

对策：让反向请求回调只做本地工作（调 LLM、弹窗问用户），不要在回调里回头调用同一个 session；确实需要时，把后续调用挪到回调之外去做。

11.8.3 反模式三：硬编码 sampling 回调

# ❌ 回调里直接调 OpenAI，忽略用户的实际 LLM
async def sampling_cb(ctx, params):
    return await openai.chat.create(...)

# ✅ 根据配置路由到用户选择的 LLM
async def sampling_cb(ctx, params):
    provider = get_user_llm_provider()
    return await provider.complete(params)

11.8.4 反模式四：不处理 Server 重启

现象：MCP Server 重启后原 session 失效，但 Agent 不知道，继续用旧 session 发请求只会失败（stdio 下读流关闭时，挂起中的请求会收到 CONNECTION_CLOSED（-32000）错误，见 src/mcp/shared/session.py:444-457）。

对策：

实现心跳检测（send_ping()，记得配超时）
捕获 session 失效错误，自动重连
用了 ClientSessionGroup 的话，先 disconnect_from_server() 摘掉失效的 session，再 connect_to_server() 重连——Group 本身不提供断连通知

11.9 源码账本：client/ 目录与核心类

11.9.1 `mcp/client/` 目录的真实结构

mcp-python-sdk 的 src/mcp/client/ 实测 4517 行（其中 auth/ 子目录 1500 行、其余 3017 行）——主要文件如下：

文件 / 子目录	行	角色
`session.py`	480	`ClientSession` 单服务器会话（章节主线）
`session_group.py`	410	`ClientSessionGroup` 多服务器聚合
`client.py`	308	`Client` 高层 wrapper（dataclass、章节没明确介绍）
`streamable_http.py`	588	Streamable HTTP transport（最复杂）
`stdio.py`	270	stdio transport
`sse.py`	160	SSE transport
`websocket.py`	85	WebSocket transport
`_memory.py`	78	InMemoryTransport（仅供测试 / Client 直连 server 用）
`_transport.py`	21	Transport ABC base 接口
`context.py`	16	client-side request context
`src/mcp/client/auth/oauth2.py`	646	整个 client 最大文件——OAuth 2.1 完整实现
`src/mcp/client/auth/utils.py`	339	OAuth 工具（PKCE、metadata 发现等）
`src/mcp/client/auth/extensions/client_credentials.py`	485	客户端凭据 grant type 扩展

两个值得注意的尺寸事实——

1. src/mcp/client/auth/ 整个子目录 1500 行（oauth2.py 646 + utils.py 339 + extensions/client_credentials.py 485 + 少量 __init__ / exceptions）——比 session + session_group 加起来还多（480+410=890）。Python SDK 在 OAuth 上的投入和 TS SDK 一样夸张——和本专栏第 9 章 §9.11 提到的 TS SDK auth.ts 1745 行相互印证：完整的 OAuth 2.1 客户端实现，在两个 SDK 里都是四位数行数的量级。

2. streamable_http.py 588 行 vs stdio.py 270 行 vs sse.py 160 行——Streamable HTTP 是最复杂的 transport。原因是它要支持：(a) JSON 和 SSE 两种响应模式自适应；(b) Last-Event-ID 重连重放；(c) DELETE 主动关 session；(d) 普通 POST + GET-for-stream 两种模式。四类功能叠加导致代码量是 stdio 的 2 倍多。

11.9.2 `Client` 高层 wrapper（client.py 308 行）

章节的 ClientSession 是底层 API、需要手动管理 transport——Client 是 dataclass-based 的高层 wrapper：

# 节选自 src/mcp/client/client.py:36-142（省略 docstring、私有字段与方法体）
@dataclass
class Client:
    server: Server[Any] | MCPServer | Transport | str  # ← 自动识别 4 种入参类型
    _: KW_ONLY                                         # 之后的字段只能按关键字传
    raise_exceptions: bool = False
    read_timeout_seconds: float | None = None
    sampling_callback: SamplingFnT | None = None
    list_roots_callback: ListRootsFnT | None = None
    logging_callback: LoggingFnT | None = None
    message_handler: MessageHandlerFnT | None = None
    client_info: Implementation | None = None
    elicitation_callback: ElicitationFnT | None = None

    async def __aenter__(self) -> Client: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...

server 字段的多态识别——__post_init__ 里——

isinstance(server, Server | MCPServer) → 包成 InMemoryTransport（测试 / 嵌入用）
isinstance(server, str) → URL → streamable_http_client transport
isinstance(server, Transport) → 直接用

这种"四 in 一"构造——让最常见的两种用法（连 URL / 直接嵌 server 测试）一行代码搞定：

# 测试嵌入
async with Client(my_mcp_server) as client:
    await client.call_tool("add", {"a": 1, "b": 2})

# 生产 HTTP
async with Client("https://api.example.com/mcp") as client: ...

# 自定义 transport
async with Client(my_stdio_transport) as client: ...

Client vs ClientSession 的关系——

ClientSession —— 协议层、必须自己拿到 read/write streams
Client —— 应用层、自动管理 transport + AsyncExitStack
绝大多数用户应该用 Client、不用 ClientSession —— 章节正文偏向后者是因为讲底层架构、但生产代码默认 Client

对照 TS SDK——TS 没有等价的"四态构造"Client——TS 用户必须先 new Client() 再 client.connect(transport)。Python 的 dataclass + __post_init__ 让这种多态构造极简洁——和 §10.10.8 讨论过的 Tool.from_function 是同源思路（类型推断 + Pydantic/dataclass 让 Python SDK 比 TS SDK 更紧凑）。

11.9.3 核心类的源码账本（python-sdk commit `3d7b311`，2026-04）

前面把 ClientSession、ClientSessionGroup、Client、各 transport 和 OAuth 穿了一遍。这里把真实行数与函数签名以账本形式再归拢一遍——便于读者对着源码调试时按图索骥，也便于和 TS SDK 跨章对照。所有数据取自 github.com/modelcontextprotocol/python-sdk 的 src/mcp/client/ 目录。

11.9.3.1 `session.py`（480 行）：构造函数、15 个公开方法与入站分发

方法	协议操作	关键 kwargs	返回
`__init__`	—	5 个 callback（sampling / elicitation / list_roots / logging / message_handler）+ `read_timeout_seconds` + `client_info` + kwargs-only 的 `sampling_capabilities`、`experimental_task_handlers`	—
`initialize()`	`initialize`	无	`InitializeResult`
`send_ping(*, meta=None)`	`ping`	`meta`	`EmptyResult`
`list_tools(*, params=None)`	`tools/list`	`PaginatedRequestParams`（分页游标）	`ListToolsResult`
`list_resource_templates(*, params=None)`	`resources/templates/list`	分页	`ListResourceTemplatesResult`
`set_logging_level(level, *, meta=None)`	`logging/setLevel`	`meta`	`EmptyResult`
`send_roots_list_changed()`	`notifications/roots/list_changed`	—	`None`
`call_tool(name, arguments, read_timeout_seconds, progress_callback, *, meta=None)`	`tools/call`	`progress_callback` 支持流式进度	`CallToolResult`
`list_resources(*, params=None)`	`resources/list`	同上	`ListResourcesResult`
`read_resource(uri, *, meta=None)`	`resources/read`	`meta`	`ReadResourceResult`
`subscribe_resource(uri, *, meta=None)`	`resources/subscribe`	`meta`	`EmptyResult`
`unsubscribe_resource(uri, *, meta=None)`	`resources/unsubscribe`	`meta`	`EmptyResult`
`list_prompts(*, params=None)`	`prompts/list`	分页	`ListPromptsResult`
`get_prompt(name, arguments, *, meta=None)`	`prompts/get`	`arguments: dict[str, str]`	`GetPromptResult`
`complete(ref, argument, context_arguments=None)`	`completion/complete`	`context_arguments` 对应 2025-06-18 版 schema 新增的 `context` 字段——用于多级补全场景	`CompleteResult`
`send_progress_notification(token, progress, total, message, *, meta=None)`	`notifications/progress`	`message` 是 2025-03-26 版 schema 新增	`None`
`_received_request(responder)`	入站分发	`match/case` 派发 `CreateMessageRequest` / `ElicitRequest` / `ListRootsRequest` / `PingRequest`	—

三个容易被忽视的签名细节——

call_tool 的 progress_callback 不是 kwargs-only——位置参数可以传（* 之后只有 meta）。新代码推荐用 kwargs 写法，避免把 read_timeout_seconds 与 progress_callback 的位置弄混。
complete(...) 的 context_arguments 是 2025-06-18 版 schema 增补的 context.arguments——用来在 prompt argument 自动补全时传"已填字段"作为上下文，让服务器算出更精准的候选。按更早版本实现的服务器不认识这个字段。
experimental_task_handlers（__init__ 的 kwargs-only 参数）——由 SEP-1686（PR #1645，2025-11 并入 main）引入，承载 tasks 协议扩展（对应 2025-11-25 规范里 basic/utilities/tasks 那一节），现在先走 experimental 命名空间。2026-07-28 版规范已把 Tasks 移出核心、改为官方扩展（SEP-2663）。

11.9.3.2 `session_group.py`（410 行）的函数与行号对照

成员	行号	职责
`ServerParameters` TypeAlias	67	`Stdio \| Sse \| StreamableHttp` 三选一
`_ComponentNames`（`BaseModel` 内部类）	104	反向索引：某 session 注册了哪些 tool/resource/prompt
`_session_exit_stacks: dict[ClientSession, AsyncExitStack]`	120	每会话一个 stack（第二级）
`__init__`	128-148	可选外部 `exit_stack` + `component_name_hook`
`__aenter__`	150-154	只 enter 自己创建的 exit_stack 并返回 `self`，真工作在 `connect_to_server`
`__aexit__`	156-171	先关主 stack，再用 anyio task group 并发关各会话 stack（默认用法下后者是空操作，见 §11.3.1）
`call_tool`	193-211	按外部名查 session、用内部原名转发
`disconnect_from_server`	213-246	单 session 热拔
`connect_with_session` / `connect_to_server`	248-262	聚合已有会话 / 新建会话并聚合
`_establish_session`	264-330	按 `ServerParameters` 类型分发到对应 transport
`_aggregate_components`	332-405	临时字典 + 冲突检测 + 原子合并
`_component_name`	407-410	应用 `component_name_hook`（没设就原样返回）

_aggregate_components 长达 70 多行——它把"收 prompts / 收 resources / 收 tools / 检测冲突 / 合并 / 更新反向索引"六个阶段串在一起。临时字典不是写着好看的：冲突检测抛错时，Group 的聚合字典必须原封不动（拉取阶段的 MCPError 则只记 warning、不回滚，见 §11.3.2）。

11.9.3.3 `client/` 目录总行数分布

文件	行	说明
`session.py`	480	主文件
`session_group.py`	410	多服务器聚合
`client.py`	308	高层 `Client` wrapper
`streamable_http.py`	588	最大的 transport——SSE 模式自适应 + Last-Event-ID 重连
`stdio.py`	270	Unix/Windows 分支 + 三段 shutdown
`src/mcp/client/auth/oauth2.py`	646	单文件最大
`src/mcp/client/auth/utils.py`	339	PKCE / metadata 发现工具
`src/mcp/client/auth/extensions/client_credentials.py`	485	`client_credentials` grant type 扩展
`sse.py`	160	最老的 HTTP transport
`websocket.py`	85	Python 独有全双工
`_memory.py`	78	InMemoryTransport（测试和 `Client(server)` 直连用）
`_transport.py`	21	抽象基类
`context.py`	16	client-side `RequestContext` 薄封装

三条可以拿来和 TS SDK 互校的结论——

Python auth/ 子目录总计 1500 行（oauth2.py 646 + utils.py 339 + client_credentials.py 485 + 零散文件）；TS SDK 的 auth.ts 单文件就 1745 行（本专栏第 9 章 §9.11 实测）。Python 拆成多个文件、TS 挤在一个文件里，总量是同一量级。
streamable_http.py 588 行是最大 transport——约是 sse.py（160）的 3.7 倍、stdio.py（270）的 2.2 倍。复杂度主要来自三件事：（a）响应体 JSON vs SSE 自适应；（b）Last-Event-ID 断线续传；（c）DELETE 主动销毁 session。本专栏第 13 章会把这三件事的工程细节展开讲。
_memory.py 才 78 行——它是 Client(server) 构造器里"直接传 Server 实例"那一路用的。对应 TS SDK 里的 InMemoryTransport——TS 版本用来写单元测试很方便，Python 版本则进一步被 Client 吸收成"零配置嵌入模式"。

11.9.4 Transport 默认值与超时表（实测源码）

transport 层散落着许多默认值，这些值一旦误解会直接导致生产故障。这里把最重要的 11 个默认值做成账本，全部基于 src/mcp/client/ 的现行代码。

参数	值	出处	触发条件
`stdio_client` 内存流 buffer size	0	`src/mcp/client/stdio.py:115-116`	每次 send 都必须有 receiver 就绪，天然背压，慢消费者会拖慢生产者
`PROCESS_TERMINATION_TIMEOUT`	2.0 秒	`src/mcp/client/stdio.py:48`	关 stdin 后等子进程自行退出的窗口；超时后调 `_terminate_process_tree()`
`_terminate_process_tree(timeout_seconds=…)`	2.0 秒	`src/mcp/client/stdio.py:255`、实现在 `src/mcp/os/posix/utilities.py`	SIGTERM 之后再等这么久；仍不退出就 SIGKILL。注意与上一行是两个独立的 2 秒窗口，不是同一个
`stdio_client(errlog=...)` 默认	`sys.stderr`	`src/mcp/client/stdio.py:105-106`	子进程 stderr 直接冒到父进程 stderr——调试方便，生产里可能要重定向到日志文件
stdio 启动目录 `cwd` 默认	`None`（继承父进程）	`StdioServerParameters` 定义	生产里最好显式设——避免把 server 绑到 Agent 的工作目录
stdio 环境变量 `env` 默认	`get_default_environment()` 白名单	`src/mcp/client/stdio.py:51-68`、`:125`	不是继承全部 env——只转发 `PATH / HOME / USER` 等安全白名单，防止 `ANTHROPIC_API_KEY` 等泄漏给子进程；显式传了 `env` 时是"白名单 + 你给的"合并
`streamable_http_client.terminate_on_close`	True	`src/mcp/client/streamable_http.py:523`	退出上下文时向 server URL 发 `DELETE`——只在 server 下发过 session id 时才发（同文件 `:586`），无状态 server 不下发 id，本来就不会发
`MCP_SESSION_ID` header	`mcp-session-id`	`src/mcp/client/streamable_http.py:44`	代码里是小写字面量，但从响应里读它走的是 httpx 的 `Headers.get()`，大小写不敏感——反代把它改写成 `Mcp-Session-Id` 不影响
POST 响应 content-type 判别	`application/json` / `text/event-stream`	`src/mcp/client/streamable_http.py:290-301`（`_handle_post_request` 内）	前者走 JSON 分支，后者走 `_handle_sse_response`；两者都不是就给这次请求回一条 `INVALID_REQUEST` 错误——反代若改写了 `Content-Type`，请求会直接失败
`ClientSession.read_timeout_seconds`	`None`（无超时）	`session.py:__init__`	不设超时的 send_request 会永久挂起——生产必须设，比如 60s
`OAuthClientProvider.async_auth_flow` 401 → 完整重新授权	1 次	`src/mcp/client/auth/oauth2.py`	401 触发完整 OAuth 流程后带新 token 重试一次；刷新 token 是前置动作，不在 401 分支——避免无限循环

11.9.4.1 `get_default_environment()` 白名单为什么重要

stdio.py 不把父进程全部 os.environ 塞给子进程、而是默认白名单过滤——非 Windows 平台（Linux / macOS）的白名单是 HOME、LOGNAME、PATH、SHELL、TERM、USER（src/mcp/client/stdio.py:28-45，Windows 另有一份），其他一律丢弃。

这是一次有意识的安全设计——子进程是第三方 MCP server、可能是不信任代码。如果把 ANTHROPIC_API_KEY / OPENAI_API_KEY / AWS_ACCESS_KEY_ID 等全部透传过去、server 一行 os.environ.get("ANTHROPIC_API_KEY") 就拿走你的密钥。TS SDK 的 StdioClientTransport 也做了同样的过滤——packages/client/src/client/stdio.ts 里同名的 DEFAULT_INHERITED_ENV_VARS 与 getDefaultEnvironment()——这是两个 SDK 难得完全一致的安全默认。

需要给子 server 传敏感凭据时，显式写 env={"MY_TOKEN": "xxx"}——明确同意而不是默认透传。

11.9.4.2 `read_timeout_seconds` 不设值就是永久挂起

ClientSession 的 read_timeout_seconds 默认 None 这件事每个生产项目都会踩一次——

本地开发测试：server 一切正常、call_tool 毫秒级返回、没人注意超时
部署到生产：某个 server 某次调用慢了或死锁、await session.call_tool(...) 永远不返回
Agent 主循环卡死、整个 user request 挂在 HTTP gateway 上

对策——

初始化时显式传 ClientSession(..., read_timeout_seconds=60)——全局兜底
单次调用临时放宽/收紧：session.call_tool(name, args, read_timeout_seconds=5)——快工具严查、慢工具放宽
不要写 read_timeout_seconds=None 这种"显式永久等"——要么是 bug、要么加注释说明为什么刻意这么做

还要知道超时之后发生什么：send_request 只在本地抛 MCPError(code=REQUEST_TIMEOUT)（src/mcp/shared/session.py:291-297），不会给 server 发 notifications/cancelled——server 那边的这次调用仍会继续跑完。需要让 server 停手，得自己补发取消通知。

和本专栏第 4 章 §4.8「超时与重连」相互印证——超时不是一个值、是一组值（transport 级 / session 级 / 单次调用级）逐层覆盖。

11.10 消息分发与接收循环

11.10.1 `_received_request` 的 match/case 分发路径全图

session.py 的 _received_request 是 Client 端最密集的状态机——它接收服务器反向发来的所有请求，按类型路由到对应 callback。下面把 match/case 显式处理的入站请求列出来，并标明每个分支的默认行为（tasks 相关的 4 种见 §11.10.1.1）。

入站请求类型	协议名	默认回调名	未注册时行为	典型用途
`CreateMessageRequest`	`sampling/createMessage`	`_default_sampling_callback`	返回 `ErrorData(INVALID_REQUEST, "Sampling not supported")`	第 17 章的 sampling 协议——server 借 client 的 LLM
`ElicitRequest`	`elicitation/create`	`_default_elicitation_callback`	返回 `ErrorData(..., "Elicitation not supported")`	第 18 章——server 向用户发起结构化提问
`ListRootsRequest`	`roots/list`	`_default_list_roots_callback`	返回 `ErrorData(..., "List roots not supported")`	第 18 章——server 请求当前工作空间列表
`PingRequest`	`ping`	内置 `EmptyResult()`	直接应答、不经 user 代码	第 4 章讨论过的心跳 / 保活

11.10.1.1 协议规定的入站请求

2025-11-25 版 schema 的 ServerRequest 联合类型里共有 8 种 server → client 的 request：上表 4 种，外加 tasks 相关的 tasks/get、tasks/result、tasks/list、tasks/cancel（schema/2025-11-25/schema.ts 的 ServerRequest）。后 4 种由下面说的前置分发处理，所以 _received_request 的 match/case 只需要显式覆盖 4 个 case。

match 之前还有一段前置分发：if self._task_handlers.handles_request(...) 会把 tasks 扩展相关的请求先截走；CreateMessageRequest / ElicitRequest 分支里也会检查 params.task is not None，是任务增强请求就交给 _task_handlers.augmented_sampling / augmented_elicitation，否则才走用户注册的 callback。

实际代码里的 case _: 是 pass（源码是 pass # Task requests handled above by _task_handlers）——但它基本走不到：协议外的 method 在更早的一步就被拦下了。_receive_loop 先用 ServerRequest 的 TypeAdapter 校验入站请求，校验不过就进兜底分支，回一条 -32602 "Invalid request parameters"（src/mcp/shared/session.py:357-388）。实测 server 发 foo/bar，client 回的正是这条——不是 METHOD_NOT_FOUND（-32601），所以 server 侧别指望靠 -32601 判断"client 不支持这个方法"。

11.10.1.2 默认回调的错误消息：约定，而非协议

仔细看四个 _default_*_callback 的错误消息——

"Sampling not supported"
"Elicitation not supported"
"List roots not supported"

都是 <功能名> not supported 格式，错误码统一是 INVALID_REQUEST（-32600）。这只是 SDK 内部的措辞一致性，不是协议约定——规范并没有为"能力未声明"定义专门的错误码或消息格式，所以 server 侧靠字符串匹配 "not supported" 来分流是不可靠的做法。正确的判断依据是握手阶段拿到的 ClientCapabilities：客户端没声明 sampling，server 就不该发 sampling/createMessage。本专栏第 3 章 §3.4 讨论过 JSON-RPC 错误码在 MCP 里的用法，这里是一个"约定先于规范"的典型。

11.10.1.3 `with responder:` 为什么是同步上下文管理器

读源码会注意到——match 里每个 case 都写 with responder:（同步 with、不是 async with）。这是因为 RequestResponder.__enter__ 只做同步的 CancelScope 设置、不需要异步资源：

__enter__ — 创建 anyio.CancelScope 并 enter；收到对应的 notifications/cancelled 时，responder.cancel() 会取消这个 scope
__exit__ — exit scope；如果请求已应答，就把这个 responder 从 session 的 _in_flight 表里移除

设计意图是：client 处理 server 发来的 sampling/createMessage（调 LLM 可能要 30 秒）时，server 中途发 notifications/cancelled，session 调 responder.cancel() 取消 scope，让正在 await 的 LLM 调用立刻抛 CancelledError，同时回一条 code: 0, "Request cancelled" 的错误响应（src/mcp/shared/session.py:142-155）。

但在基准版里，这条路对 _received_request 里当场处理的请求走不通。 原因见 §11.8.2：接收循环是直接 await self._received_request(responder) 的，回调没跑完，后面的 notifications/cancelled 根本还没被读出来；等读到时，responder 已经应答、已从 _in_flight 移除，取消落空。实测 server 在 0.2 秒时发 cancelled，2 秒的 sampling 回调照样跑完，client 照样回了正常 result。只有没在 _received_request 里当场应答、被转交给 message_handler、再由用户代码另起任务处理的请求，cancelled 才可能在它还在飞行中时到达并生效（第 3 章 §3.8 讨论过 cancellation 的 wire 格式）。

11.10.2 ClientSession 的消息处理循环（`_receive_loop` 与 `_received_notification`）

前面讲了"发请求" call_tool 和"收请求" _received_request——但 MCP 还有第三种消息：server 发来的 notification（通知、无需应答）。处理入口是 _received_notification。

11.10.2.1 客户端关心的 notification 类型

Server → Client notification	含义	默认 callback
`notifications/message`	结构化日志	`_default_logging_callback`（no-op）
`notifications/resources/updated`	某 resource 内容变了	传给 user `message_handler`
`notifications/resources/list_changed`	资源列表变更	同上
`notifications/tools/list_changed`	工具列表变更	同上
`notifications/prompts/list_changed`	提示词列表变更	同上
`notifications/progress`	长耗时操作进度	路由到对应 `progress_callback`
`notifications/cancelled`	server 主动取消	取消对应请求的 CancelScope（受串行接收所限，见 §11.10.1.3）

11.10.2.2 `message_handler` 的兜底地位

注意——三个 *_list_changed 通知和 resources/updated 都没有专门 callback。它们只会路由到 message_handler（ClientSession.__init__ 的参数）——一个"什么都收"的兜底函数。

这个设计反映了 SDK 的取舍——high-churn 的通知类型用一个通用 handler 处理、避免给每种 list_changed 都单独配 callback 造成 API 表面爆炸。缺点是—— user code 要写 isinstance(msg, ResourceListChangedNotification) 之类的 dispatch。

推荐实践——

async def handler(msg):
    match msg:
        case types.ResourceListChangedNotification():
            await refresh_resource_cache()
        case types.ToolListChangedNotification():
            await refresh_tool_cache()  # 会顺带重算 _tool_output_schemas
        case types.LoggingMessageNotification(params=p):
            logger.info("server log [%s]: %s", p.level, p.data)  # p.level 是字符串，不能直接传给 logger.log

11.10.2.3 `_tool_output_schemas` 的失效时机

工具列表变更时、本章 §11.2.2 讲的 schema 缓存必须失效——否则 call_tool 会拿旧 schema 校验新 tool 的返回、误报 validation 错误。

查 src/mcp/client/session.py:393-409 的源码，list_tools() 返回时做的是逐条 upsert，而不是整体重建：

# Cache tool output schemas for future validation
# Note: don't clear the cache, as we may be using a cursor
for tool in result.tools:
    self._tool_output_schemas[tool.name] = tool.output_schema

源码注释解释了原因——分页拉取时 list_tools() 会被调用多次、每次只返回一页，清空缓存会把上一页的 schema 也抹掉。代价是缓存只增不减：服务器删掉某个工具后，它的旧 schema 会永远留在字典里；而且如果 user 收到 tools/list_changed 不主动调 list_tools()、被改过的工具也还是旧 schema。

对策——在 message_handler 里看到 ToolListChangedNotification 立刻调 session.list_tools()——既更新了 UI / Agent 端能看到的工具列表、又刷新了 schema 缓存。源码里这是个"二合一"操作、不要分开想。

11.10.3 一个容易被忽略的边界：`_receive_loop` 的异常传播

最后补一个源码级细节——BaseSession._receive_loop（定义在父类所在的 src/mcp/shared/session.py:351，客户端用的就是它）接到来自 read_stream 的 Exception 对象时——不会抛出，而是转发给 message_handler。

回看 ClientSession.__init__ 的签名——read_stream: ReadStream[SessionMessage | Exception]——流里可以流 Exception。这是 anyio 的惯用法——transport 层（stdio 子进程死了 / HTTP 连接断了 / JSON 解析失败）把 exception 包成 stream 元素、而不是抛 exception 弄死 receive loop。

这对 user 的影响——

写 message_handler 时记得判断 if isinstance(msg, Exception): handle_err(msg); return
不这么判断的话、transport 出错是静默的——stream 可能已空、但 user 不知道为什么 session 突然不工作了
推荐每个 handler 最顶上 4 行模板：

async def handler(msg):
    if isinstance(msg, Exception):
        logger.error("Transport error: %r", msg)
        await trigger_reconnect()
        return
    # ... 正常处理

这是 Python SDK 和 TS SDK 的第 10 个隐藏差异——TS SDK 的 transport 错误走 transport.onerror(err) 回调；Python 用 stream 元素、错误和正常消息走同一条通道。Python 的好处是 error 处理和消息处理都走 match/case、代码路径唯一。但两边的默认值一样安静：TS 的 Protocol._onerror 只调用可选的 this.onerror?.(error)（mcp-ts-sdk/packages/core/src/shared/protocol.ts:524-525），用户没设 onerror 时错误同样被吞掉。

11.11 OAuth2 的五个关键决策点

§11.5 给出了 PKCEParameters / TokenStorage / OAuthClientProvider 三个构件——这里把它们串起来、看一次完整授权的五个决策点。这些决策点决定了生产环境下的使用体验、也是和 TS SDK 对照时最容易混淆的地方。

11.11.1 决策一：token 从哪里来（TokenStorage 的读取顺序）

async_auth_flow 每次进入先看 self.context.current_tokens——内存中的 OAuthContext 状态。它从哪来？第一次进入时由 _initialize() 从 TokenStorage.get_tokens() 异步读一次，读完置 self._initialized = True；此后只有刷新失败把 _initialized 置回 False 时才会在下一次请求里重读（src/mcp/client/auth/oauth2.py:476-480、:520-522）。

这意味着——

如果 token 在外部被刷新（比如另一个 client 实例更新了 Keychain）、当前 provider 通常不会感知——正常路径上它只读一次。
自己换来的 token 会在 _handle_token_response 里同时更新 context.current_tokens 并调 TokenStorage.set_tokens(...) 回写——读一次、写多次的模型。
对多 client 竞争的场景、TokenStorage 实现应该做带版本号的乐观锁或每次都 get_tokens() 重读——SDK 默认不这么做、生产需要自己扩展。

11.11.2 决策二：401 之后完整重新授权，只重试一次

async_auth_flow 收到 401 后走的是完整重新授权（PRM 发现 → OASM 发现 → scope 选择 → 客户端注册 → 授权码换 token），不是简单刷新——刷新是前置动作，在发请求之前就做了。完整流程走完只 _add_auth_header 重试一次，再失败就把响应交回上层——不是无限循环。

这样做避免了死循环——如果重新授权拿到的 token 仍然被拒（比如 server 端的受众或 scope 配置有误），再走一遍也是 401，循环只会堆请求。SDK 选择"快速失败"、把响应交给 user code 决定下一步。

11.11.3 决策三：PKCE 的 code_verifier 长度用上限

PKCEParameters.generate() 用 range(128) 生成 code_verifier——直接取 RFC 7636 允许的最大长度。

最短只要 43 字符就合规（RFC 7636 §4.1）——为什么用 128？直接取上限是最省事的保守做法：字符集 [A-Za-z0-9-._~] 共 66 个字符，log2(66^128) ≈ 774 bit 熵，远超 AES-256 的 256 bit——对 authorization code 截获后爆破 verifier 的攻击者来说，几乎不可能穷举。

代价只是 token 请求体里的 code_verifier 比最短合规长度多 85 字符；授权 URL 里的 code_challenge 是 SHA-256 摘要的 base64url，固定 43 字符，不受 verifier 长度影响。

11.11.4 决策四：authorization code 交换用 POST form-urlencoded

实际查 OAuth 2.1 spec 和 src/mcp/client/auth/oauth2.py 实现——token endpoint 请求 body 是 application/x-www-form-urlencoded、不是 JSON。这是 OAuth 社区的历史遗留——2012 年 RFC 6749 定的、现在改不动了。

新手常犯的错——把 grant_type / code / code_verifier 塞进 JSON body 发给 token endpoint——严格按规范实现的 OAuth server 会拒绝这种请求。SDK 用的是 httpx 的 data=...（form-urlencoded），还显式带上 Content-Type: application/x-www-form-urlencoded（src/mcp/client/auth/oauth2.py:379-408）。

11.11.5 决策五：Dynamic Client Registration 的 fallback

注意这里的规范口径在 2025-11-25 版有过调整。该版 basic/authorization.mdx 明确写的是：授权服务器与 MCP 客户端 SHOULD 支持 OAuth Client ID Metadata Documents（CIMD，draft-ietf-oauth-client-id-metadata-document-00），而 RFC 7591 的 Dynamic Client Registration 只是 MAY。规范给出的客户端优先级是：先用预配置的 client ID；授权服务器元数据里有 client_id_metadata_document_supported 就用 CIMD；DCR 只作为回退（靠 registration_endpoint 判断是否可用）。本专栏第 15 章 §15.3「客户端注册的三种策略」讲的正是这三条路径。

Python SDK 的实现顺序与规范一致——预配置的 client 信息最先生效：_initialize() 已经从 TokenStorage.get_client_info() 读过一次，读到了就整个跳过注册。src/mcp/client/auth/oauth2.py 里 401 之后的第 4 步只在"还没有 client_info"时才执行：授权服务器声明支持且本地配了 client_metadata_url 就用 CIMD，否则回退 DCR——registration_endpoint 缺失时还会退到 <授权服务器基址>/register 去试（src/mcp/client/auth/utils.py:218-230），注册失败则抛 OAuthRegistrationError，不会再有别的兜底。

11.12 跨章校对：Python Client 在 MCP 专栏里的位置

本章内容不是孤立的。Python Client 的每一块实现都能在本专栏其他章找到对应的协议规范或服务端镜像。下面这张表把五对互查点列出来——读者发现本章某段描述和其他章有出入时，以规范原文和本专栏 pin 的源码为准。

主题	本章位置	协议规范章	Server 镜像 / 兄弟 SDK
`initialize` 能力协商	§11.2.1	第 4 章 §4.2 初始化握手详解、§4.4 能力声明机制	第 10 章 §10.6 会话管理与初始化握手
`tools/call` + output schema 校验	§11.2.2 `_validate_tool_result`	第 5 章 §5.2 Tool 的定义结构	第 10 章 §10.3 Pydantic 自动生成 schema
`resources/list` + `resources/subscribe`	§11.2.2 签名表	第 6 章 §6.4 资源发现、§6.5 订阅与实时更新	第 10 章 §10.2 `@resource` 装饰器
`CreateMessageRequest`（sampling 反向调用）	§11.2.3 `_received_request` 分发	第 17 章 §17.3 CreateMessage 请求与响应	第 9 章 §9.6 请求处理器：Server 发起的请求
stdio transport graceful shutdown	§11.4.1 三段式关闭	第 12 章 §12.4 客户端实现：进程管理的全生命周期	第 9 章 §9.5 Transport 选择与连接策略
Streamable HTTP `mcp-session-id`	§11.4.2 + §11.9.4 默认值表	第 13 章 §13.3 会话管理：有状态与无状态的权衡	第 10 章 §10.5 Starlette/ASGI 集成
OAuth2 PKCE + TokenStorage	§11.5	第 15 章 §15.4 授权码流程与 PKCE、§15.5 令牌管理	第 10 章 §10.5 认证中间件链
`ClientSessionGroup` 多服务器聚合	§11.3	— MCP 规范没有 group 概念，这是 SDK 层扩展	TS SDK 不提供、需 Agent 框架自行实现

11.12.1 Client 与 Server 的"镜像关系"

每个 Client 方法都有一个 Server 端处理函数镜像——

session.call_tool("add", ...) ↔ Server 端 @server.tool() 装饰的 add() 函数（见第 10 章 §10.10.10 从装饰器到协议 handler 的完整调用链）
session.subscribe_resource("file:///log") ↔ Server 端低层 Server 的 on_subscribe_resource 回调记账（示例 examples/servers/everything-server 用一个 resource_subscriptions: set[str]）+ ServerSession.send_resource_updated() 推送；高层 MCPServer 没有接订阅（第 6 章 §6.5.3）
客户端 sampling_callback ↔ Server 端 ctx.session.create_message(...) 发起的反向请求

这种对称性意味着——调试 Client bug 时、经常要同时开两端的日志对着看：在两端各自的入口处都把 Python logging 调到 DEBUG、记录每次 send/receive 的 method 与 id，再用相同的 timestamp 前缀就能在两边日志里逐条对齐同一次请求-响应往返。

11.12.2 和 TS SDK 的互查

本章 §11.6 已经列过 9 项差异。以下三条是最容易混淆的点、读者在 Python 和 TS 之间切换时要留意——

async with vs 手动 connect/close——Python 忘了 async with ClientSession，接收循环根本不会启动、initialize() 当场挂起，问题立刻暴露；TS 用户忘了 close() 则会静默泄漏 socket 或子进程。本专栏第 9 章给出了 TS 的 try/finally 最佳实践。
list_tools() 的返回类型——Python 返回 ListToolsResult（驼峰字段反序列化为 snake_case：next_cursor）；TS 返回 { tools, nextCursor }（camelCase）。混用两端代码时的 JSON schema 别名要留心。
分页参数位置——Python 放在 kwargs-only 的 params=PaginatedRequestParams(cursor=...)；TS 直接第二参数传 { cursor: "..." }。TS 更紧凑、Python 更类型安全——没有哪个更好。

11.12.3 ClientSessionGroup 是"SDK 层创新"而非协议规范

特别要指出的是——ClientSessionGroup 不是 MCP 规范的一部分。规范只讲单 client 单 server 的协议；多服务器聚合是 Python SDK 提供的上层工具。这带来两个工程后果——

好：换了 SDK 版本、ClientSessionGroup API 可能变、协议本身不会变——底层 ClientSession 始终稳定。
需要注意：其他语言的 SDK 不一定有对应抽象（本专栏对照的 TS SDK 就没有）——跨语言 Agent 框架得自己实现聚合层，各家实现解决的是同一个问题、但没有统一接口。

这也揭示了 MCP 生态下一步的可能走向——"多服务器聚合层"是否要进规范？截至 2026-07-28 版规范，changelog 里没有相关条目，它仍留在 SDK 侧。

11.13 本章小结

本章深入分析了 MCP Python SDK 的客户端实现。

核心要点

ClientSession 作为单服务器会话的核心抽象，继承自 BaseSession 并提供了完整的 MCP 协议操作方法
ClientSessionGroup 在此基础上实现了多服务器管理与组件聚合
Python SDK 充分利用了 anyio 的结构化并发、async with 上下文管理器、Pydantic 类型校验等语言特性
传输层的四种实现（stdio、SSE、Streamable HTTP、WebSocket）通过统一的读写流抽象与上下文管理器模式，实现了对应用层的完全透明
OAuth2 认证作为可插拔的 httpx 中间件，为 HTTP 类传输提供了标准化的安全方案

记住这三个设计范式

async with 贯穿一切——资源获取与释放绑定在语法结构上
Protocol + TypeAdapter——结构化子类型 + 运行时校验的组合拳
聚合 + 命名空间 + 反向索引——ClientSessionGroup 的设计范式可以迁移到其他需要"聚合多个同类服务"的场景

下一章我们将进入传输层的深度实现——第 12 章详解 stdio 传输层，看子进程通信的精细工程。

进入第 12 章之前：带着这三个问题去读

读完本章应该能在脑子里复现 ClientSession 从构造到 async with 退出的完整路径。第 12 章会把视角从会话层进一步下沉到字节层——以下三个问题可以作为阅读指针：

Server 往 stdout 混进一行非 JSON 日志时，stdio_client 的 stdout_reader 怎么处理？和 TS SDK 的做法有什么不同？
Windows 下 start_new_session=True 不可用——SDK 用 Job Object 替代实现进程树整体终止、这个技术细节本章只提了名字、第 12 章会展开源码。
三段式 shutdown 的两个超时都是 2.0 秒（PROCESS_TERMINATION_TIMEOUT 与 _terminate_process_tree(timeout_seconds=2.0)）——TS SDK 的 close() 用的是什么时序？第 12 章给出对照。