MCP Python SDK 客户端传输层全解：Streamable HTTP、stdio、内存直连与 SSE

MCP Python SDK 客户端传输层全解Streamable HTTP、stdio、内存直连与 SSE【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk本篇文章围绕官方 Python SDKpython-sdk中客户端Client与服务器通信所依赖的**传输层transport**展开逐一讲解 Streamable HTTP、stdio 子进程、进程内in-memory与旧版 SSE 四种传输方式的用法、底层实现与注意事项。读完本文你将掌握如何根据部署形态选择传输方式、如何自定义httpx2.AsyncClient配置认证与超时、如何理解重定向与子进程环境隔离策略以及为什么可以编写属于自己的自定义传输。传输层是什么Client如何自动选择通信方式在 MCP 协议里客户端与服务器之间的消息传递依赖一个传输transport——它是真正负责搬运消息的载体。官方 SDK 的设计原则是你不需要单独配置传输因为Client只接收一个位置参数并会根据该参数的类型自动推断出对应的传输方式详见 src/mcp/client/client.py。从源码结构看Client.__post_init__会根据server参数的形态解析出对应的连接器connector传入strURL→ 自动包装为streamable_http_client(url)传入StdioServerParameters→ 自动包装为stdio_client(params)传入服务器对象MCPServer/Server→ 在进程内直接建立连接内存传输传入其他对象 → 直接当作传输transport本身进入上下文。这些连接逻辑定义在 src/mcp/client/client.py 的_connect_transport与_connect_inproc中前者把传输产出的(read, write)流对交给JSONRPCDispatcher驱动后者在legacy模式下通过InMemoryTransport走完整的 JSON-RPC 流在auto/现代协议模式下则通过DirectDispatcher对等连接直接调用。注意这里讨论的是客户端侧的传输。服务端侧mcp.run()做了什么、部署时暴露什么请参阅 运行你的服务器。Streamable HTTP生产环境的首选传输把 URL 字符串直接传给Client即可获得Streamable HTTP传输——这是官方推荐优先使用的生产级传输也是部署在 HTTP 网关/反向代理之后的标准选择from mcp import Client async def main() - None: async with Client(http://localhost:8000/mcp) as client: result await client.list_tools() print([tool.name for tool in result.tools])完整可运行示例见 docs_src/client_transports/tutorial002.py。上面这段代码就是一个完整的生产客户端。SDK 会自动把 URL 包装进streamable_http_client(...)其底层默认使用一个为 MCP 量身定制的httpx2.AsyncClient。超时默认值定义在 src/mcp/shared/_httpx_utils.pyconnect / write / pool 超时30 秒MCP_DEFAULT_TIMEOUT 30.0read读取超时300 秒MCP_DEFAULT_SSE_READ_TIMEOUT 300.0即 5 分钟——因为服务器可能长时间保持响应流打开例如订阅类长连接读取超时被刻意放宽。构造 ≠ 连接async with才是打开的时刻一个刚构造出来的Client并未连接。构造过程只负责选择传输方式真正打开传输的是async with上下文管理器。如果你在进入上下文之前就试图使用连接SDK 会明确报错RuntimeError: Client must be used within an async context manager换句话说当你写下Client(http://...)这一行时什么都没解析、没拉取、没启动——这一行是零成本的。连接的建立、协议的协商全部发生在进入async with块之后。自备httpx2.AsyncClient认证、Cookie、代理与 mTLS一旦你需要Authorization头、Cookie、代理、mTLS 或自定义超时就应当自己创建httpx2.AsyncClient并交给streamable_http_clientimport httpx2 from mcp import Client from mcp.client.streamable_http import streamable_http_client async def main() - None: async with httpx2.AsyncClient( headers{Authorization: Bearer ...}, timeouthttpx2.Timeout(30.0, read300.0), ) as http_client: transport streamable_http_client(http://localhost:8000/mcp, http_clienthttp_client) async with Client(transport) as client: result await client.list_tools() print([tool.name for tool in result.tools])完整示例见 docs_src/client_transports/tutorial003.py。使用这种方式时请务必留意两点httpx2.AsyncClient的所有权属于你因此进入与退出它的上下文是你的责任。SDK 永远不会关闭一个它自己没有创建的客户端。上面示例中http_client的async with与Client的async with嵌套顺序正是为了确保先退出 MCP 会话、再关闭 HTTP 客户端。streamable_http_client(url, http_client...)返回的是一个传输对象而Client(transport)接受任何传输对象——这和传入 URL、传入StdioServerParameters的用法完全一致只是把传输的创建权交给了你。关于 TLS 有一个值得注意的细节httpx2通过truststore校验证书——即使用操作系统信任库而不是内置的 CA 列表。在缺少可用系统 CA 存储的最小化容器环境中可以通过标准环境变量SSL_CERT_FILE/SSL_CERT_DIR指定证书或者向自己的httpx2.AsyncClient传入显式的verifyssl_context。相关背景见httpx与httpx-sse被httpx2替代。迁移警告headers与timeout参数已被移除旧版本中streamable_http_client曾直接接受headers和timeout关键字参数现在已不再接受。它的全部参数只有三个url、http_client和terminate_on_close。如果沿用旧习惯写headers会得到TypeError: streamable_http_client() got an unexpected keyword argument headers所有与 HTTP 相关的配置现在都收敛到你传入的那个httpx2.AsyncClient上可对照 src/mcp/client/streamable_http.py 的函数签名确认。基于httpx2认证、代理、重试与 OAuth 的接入点httpx2保留了大家熟悉的httpxAPI所以如果你熟悉httpx那么在这里你就已经知道如何做认证、代理、事件钩子event hooks、重试与连接数限制。SDK 在httpx2之上既不增也不减——唯一的例外是下文要讲的重定向处理。OAuth 的接入点也在这里httpx2.AsyncClient(authOAuthClientProvider(...))。完整的 OAuth 客户端流程请参阅 OAuth 客户端。重定向策略只跟随同源传输只会连接到你给它的那个 URL 对应的源origin不会去别的源。重定向规则具体如下允许跟随307/308重定向且保持相同 scheme、host、port以及同 host 下的http://→https://升级。这覆盖了最常见的/mcp→/mcp/尾斜杠重定向。拒绝跟随跳转到任何其他位置的 302 等重定向。调用会直接失败MCPError: Redirect to https://other.example.com/mcp not followed; use that URL as the endpoint if it is the intended server如果报错中给出的 URL 正是你想要的服务器就把它写进配置如果不是说明服务器本身或它前面的代理配置有误。这条规则对你传入的任何httpx2.AsyncClient都成立即使你在客户端上设置了follow_redirectsMCP 请求也不会参考它——无论开还是关。SDK 内部的 OAuth 提供方对其自身请求也采用同样的同源规则。还有一个常见的排障提示如果错误信息是Redirect to http://… not followed: it would downgrade this HTTPS endpoint to plain HTTP这通常意味着服务器位于一个它自身并不知情的 TLS 终止代理TLS-terminating proxy之后却发出了http://的重定向。解决办法是在服务端修复见 部署与扩展或者直接使用错误信息中建议的那个确切的https://…/URL。该错误信息的完整构造逻辑可查看 src/mcp/client/streamable_http.py 的_unfollowed_redirect其中对HTTPS 降级为 HTTP与跳转到其他源两种情况给出了不同的提示文案。stdio以子进程方式运行服务器stdio服务器本质上是一个子进程客户端负责启动它向它的 stdin 写入 JSON-RPC并从它的 stdout 读取 JSON-RPC。桌面宿主desktop host在本机运行服务器的方式正是如此——宿主本身就是这套代码加一个 UI。同样的关系从宿主一侧看以配置文件的形式呈现见 连接到真实宿主。用StdioServerParameters描述要启动的进程然后交给Clientfrom mcp import Client, StdioServerParameters server StdioServerParameters( commanduv, args[run, server.py], env{BOOKSHOP_API_KEY: secret}, ) async def main() - None: async with Client(server) as client: result await client.list_tools() print([tool.name for tool in result.tools])完整示例见 docs_src/client_transports/tutorial004.py。生命周期进入启动、退出清理进入async with块会启动子进程退出时会关闭它先关闭 stdin等待进程自行退出如果它迟迟不退则强制终止。整个过程由 SDK 负责你不需要手动清理。从 src/mcp/client/stdio.py 的模块说明可以看到关闭流程遵循 MCP 规范序列关闭 stdin → 等待 → 终止进程树并包裹在取消屏蔽cancellation shield内且每次等待都有上限从而保证即使调用方被取消既不会泄漏存活的服务器进程也不会挂起等待。stderr 与stdio_client的低层用法默认情况下子进程的 stderr 会直接流向你的 stderr。如果想把它重定向到别处例如写入日志文件可以自己用mcp导出的stdio_client构建传输再传给ClientClient(stdio_client(server, errloglog_file))这样做的同时你也获得了对传输更精细的控制。环境变量隔离子进程不继承你的环境子进程不会继承你的整个环境变量。它只获得一个最小化的白名单集合POSIX 平台下为HOME、LOGNAME、PATH、SHELL、TERM、USER。这样设计的目的是防止敏感信息泄漏进一个可能并非由你编写的进程中。对应实现位于 src/mcp/client/stdio.py 的DEFAULT_INHERITED_ENV_VARSWindows 平台另有独立的白名单APPDATA、HOMEDRIVE、USERPROFILE等。一个直接的后果是需要 API 密钥的服务器在白名单里找不到它。你必须通过env显式传入这些变量会叠加在白名单之上。上面示例中的BOOKSHOP_API_KEY正是这样工作的。内存传输In-Memory测试与内嵌的首选在测试场景中没有需要部署的东西也没有需要启动的进程——直接把服务器对象传进去即可from mcp import Client from mcp.server import MCPServer mcp MCPServer(Bookshop) mcp.tool() def search_books(query: str) - str: Search the catalog by title or author. return fFound 3 books matching {query!r}. async def main() - None: async with Client(mcp) as client: result await client.call_tool(search_books, {query: dune}) print(result.structured_content)完整示例见 docs_src/client_transports/tutorial001.py。这种模式下没有子进程、没有端口、没有网络上的字节。客户端与服务器是同一进程内的两个对象但调用依然会经过真实的协议层search_books会被列出、校验并被以与走 HTTP 完全一致的方式调用。这意味着内存传输的测试结果对生产环境有真实的代表性。测试 页面正是围绕这一模式构建的。同样的形态还可以作为内嵌 API使用一个自己构造服务器的应用程序可以在不经过网络跳转的情况下直接调用服务器的工具——例如在应用内部直接调用工具逻辑而不必额外起一个 HTTP 服务。SSE被 Streamable HTTP 取代的旧传输sse_client(url)位于mcp.client.sse模块是早于 Streamable HTTP 的 HTTP 传输。使用方式与其他传输一致Client(sse_client(http://localhost:8000/sse))它存在的意义是兼容仍在使用 SSE 协议的存量服务器。官方明确建议不要在它之上构建任何新东西——新项目一律使用 Streamable HTTP。Transport 协议所有传输的统一抽象对Client而言上述所有传输是同一回事。形式化的定义在 src/mcp/client/_transport.pyTransportStreams tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]] class Transport(AbstractAsyncContextManager[TransportStreams], Protocol): Protocol for MCP transports. ...传输transport就是任意一个异步上下文管理器进入后产出(read, write)一对消息流——即mcp.client中的Transport协议。Client按参数类型解析str→streamable_http_client(url)StdioServerParameters→stdio_client(params)服务器对象 → 进程内连接其他任何东西 → 直接作为传输进入。正是这最后一条规则使得stdio_client(...)、streamable_http_client(...)和sse_client(...)都能填入同一个位置——也因此你可以编写属于自己的自定义传输只要实现一个能async with并产出(read, write)流对的异步上下文管理器就可以交给Client使用从而接入任何自定义的通信管道。要点回顾Client(http://.../mcp)URL走 Streamable HTTP——生产环境的首选传输。请求头、认证、代理与超时都配置在你传入的httpx2.AsyncClient上streamable_http_client(url, http_client...)。没有headers这个关键字参数。重定向只在 URL 自身源内跟随尾斜杠307/308外加同 host 的http→https。其余一律以Redirect to … not followed失败——请在配置中写明最终 URL。stdio 就是Client(StdioServerParameters(...))。只有在需要重定向子进程 stderr 时才需要自己用stdio_client(...)包装。子进程获得的是白名单化的环境而不是你的完整环境env在其上追加变量。Client(mcp)服务器对象走内存连接适用于测试或把服务器内嵌进创建它的应用。传输就是一切能async with x as (read, write)的对象Client会把不是服务器对象、不是 URL、不是StdioServerParameters的参数直接交给该协议。构造Client只负责选择传输async with才真正打开它。当传输打开之后通信双方还需要就协议版本达成一致——通常情况下你完全不需要关心这件事当确实需要关心时请查阅 协议版本。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考