MCP Python SDK 客户端传输层完全指南Streamable HTTP、stdio、内存传输与自定义 Transport【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk导读本文围绕官方 Python SDKpython-sdkModel Context Protocol 的 Python 客户端/服务端实现中Client的传输层展开系统讲解 Streamable HTTP、stdio 子进程、内存in-process与旧版 SSE 四种传输方式的选择、配置与底层原理。读完本文你将掌握如何用一行 URL 连接生产服务器、如何通过自建httpx2.AsyncClient注入认证与代理、如何以StdioServerParameters拉起本地子进程以及为何所有传输最终都归结为同一个Transport协议。传输层是什么Client 如何根据类型推断连接方式每一个Client都是通过transport传输层与服务器对话的——传输层是真正承载消息的通道它决定了 JSON-RPC 报文以何种物理方式在客户端与服务器之间流动。在python-sdk中你永远不需要单独配置传输层。Client只接受一个位置参数server并完全根据该参数的类型推断出要使用的传输方式。这一点在源码中体现得极为直接Client的server字段类型被声明为Server[Any] | MCPServer | Transport | StdioServerParameters | str见 src/mcp/client/client.py而在__post_init__中按类型完成连接器的解析见 src/mcp/client/client.py传入Client(...)的参数类型实际使用的传输http://localhost:8000/mcpstrURLstreamable_http_client(url)StdioServerParameters(...)数据类stdio_client(params)以子进程方式运行Server(...)/MCPServer(...)对象服务器对象进程内in-process直接连接其他任意对象Transport实例直接作为传输层打开正因为这层“按类型分发”的设计stdio_client(...)、streamable_http_client(...)、sse_client(...)以及你自定义的传输都能无缝插入同一个位置。本篇文章的后续章节本质上是这张表的逐项展开。构造 ≠ 连接async with才是打开传输的时刻一个刚构造出来的Client并没有连接。构造过程只完成两件事按类型解析出传输层、准备好客户端会话真正的连接解析域名、建立 HTTP 连接、拉起子进程、完成协议握手全部发生在进入async with时。如果在进入上下文管理器之前就尝试访问连接例如访问client.sessionSDK 会明确报错RuntimeError: Client must be used within an async context manager该错误在 src/mcp/client/client.py 的session属性中抛出——_session只有在__aenter__握手成功之后才会被赋值。换句话说写下Client(http://...)这一行时什么都没发生没有解析域名、没有发起请求、没有启动子进程。这一行是“免费”的它只完成了传输层的选择。Streamable HTTP生产环境的首选传输将服务器端点 URL 以字符串形式传给Client即可获得Streamable 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])这就是完整的生产级客户端。Client内部将 URL 包装进streamable_http_client(...)其底层是基于一个按 MCP 需求配置的httpx2.AsyncClient连接connect/写入write/连接池pool超时均为 30 秒读取read超时为 300 秒——因为服务器可能长时间保持响应流打开如订阅、进度通知等场景。这组默认值定义在 src/mcp/shared/_httpx_utils.py 中并通过create_mcp_http_client()构建MCP_DEFAULT_TIMEOUT 30.0 # 一般操作秒 MCP_DEFAULT_SSE_READ_TIMEOUT 300.0 # SSE 流5 分钟秒从源码看src/mcp/client/streamable_http.pystreamable_http_client的签名是(url, *, http_clientNone, terminate_on_closeTrue)若未提供http_clientSDK 会调用create_mcp_http_client()创建一个带上述默认超时的客户端并在退出上下文时负责关闭它terminate_on_closeTrue默认值意味着退出上下文时若已建立会话传输层会向服务器发送一条DELETE 请求来终止会话见 src/mcp/client/streamable_http.py若服务器返回405说明它不支持会话终止SDK 会静默忽略。传输层还会为每次 HTTP 请求自动附加必要的 MCP 头accept: application/json, text/event-stream、content-type: application/json以及会话建立后的mcp-session-id与协商出的MCP-Protocol-Version见 src/mcp/client/streamable_http.py。自带 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])这段代码有两点需要特别注意客户端的所有权在你手中httpx2.AsyncClient是你创建的因此由你负责进入和退出它async with httpx2.AsyncClient(...)。SDK永远不会关闭一个不是它自己创建的客户端——上例中streamable_http_client只负责在传输层内部使用这个客户端其生命周期完全由外部async with管理。返回的是传输对象streamable_http_client(url, http_client...)返回一个 transportClient(transport)像接受任何其他传输一样接受它——这再次印证了上文的类型分发表。关于 TLS 的一个说明httpx2验证证书时使用的是操作系统信任库通过truststore实现而不是 SDK 内置的 CA 列表。如果运行环境没有可用的系统 CA 存储例如某些精简容器镜像有两种补救方式设置标准的SSL_CERT_FILE/SSL_CERT_DIR环境变量或为你的httpx2.AsyncClient显式传入verifyssl_context。相关背景可参阅 docs/migration.md。警惕旧习惯headers与timeout已从传输层移除警告streamable_http_client曾经直接接受headers和timeout参数现在不再支持。它的全部参数只有url、http_client和terminate_on_close。如果你按旧习惯写出streamable_http_client(url, headers...)会得到TypeError: streamable_http_client() got an unexpected keyword argument headers所有 HTTP 层面的配置请求头、超时、认证等现在都统一放在你传入的唯一一个httpx2.AsyncClient上。这也是 OAuth 接入的地方——httpx2.AsyncClient(authOAuthClientProvider(...))完整流程见 docs/client/oauth-clients.md。httpx2保留了httpx熟悉的 API如果你熟悉httpx就已经知道在这里如何做认证auth、代理proxy、事件钩子event hooks、重试retries和连接数限制connection limits。SDK 在 HTTP 层不增加也不削减任何东西——唯一的例外是下文的重定向处理。重定向策略只在同一源内跟随Streamable HTTP 传输只会连接到你给它的那个 URL且只连到该源origin。重定向的规则非常明确允许跟随保持相同 scheme、host、port 的307/308重定向同一主机上的http://→https://升级。这正好覆盖最常见的/mcp→/mcp/尾斜杠重定向。拒绝跟随指向任何其他位置的重定向一律不跟随调用会失败并报错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_redirects设置对 MCP 请求完全不被采纳正向、反向都是如此。SDK 自带的 OAuth 提供方对其自身请求也应用同样的规则。从源码看重定向判定逻辑在 src/mcp/shared/_httpx_utils.py 的_within_origin中实现比较(scheme, host, port)三元组是否完全一致或是否满足“同主机 http 默认端口 目标为 https”的升级条件。当重定向未被跟随时src/mcp/client/streamable_http.py 的_unfollowed_redirect会生成上述错误信息。实用提示如果遇到如下错误——Redirect to http://… not followed: it would downgrade this HTTPS endpoint to plain HTTP——说明服务器位于一个它自己不知道的TLS 终止代理之后且正在发出http://重定向。修复方式是在服务器端修正代理配置见 docs/run/deploy.md或者直接使用错误信息中建议的那个精确https://…/URL。stdio以子进程方式运行服务器stdio服务器本质上是一个子进程客户端负责启动它向它的 stdin 写入 JSON-RPC从它的 stdout 读取 JSON-RPC。这正是桌面主机desktop host在你机器上运行服务器的方式——一个主机本质上就是这段代码加一个界面。docs/get-started/real-host.md 从主机的视角展示了同样的关系以配置文件的形式呈现。用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])StdioServerParameters的完整字段定义在 src/mcp/client/stdio.py除示例中用到的command、args、env外还支持字段类型默认值说明commandstr必填启动服务器的可执行文件argslist[str][]传给可执行文件的命令行参数envdict[str, str] \| NoneNone额外环境变量合并在环境白名单之上cwdstr \| Path \| NoneNone启动进程时的工作目录encodingstrutf-8与服务器通信的文本编码encoding_error_handlerstrict \| ignore \| replacestrict编码错误处理方式进入async with块会启动子进程退出时会自动关闭子进程关闭 stdin、等待进程自行退出、若迟迟不退则强制终止。你永远不需要手动清理。从 src/mcp/client/stdio.py 的实现看关闭流程是先关闭读写流、给写者最多 0.5 秒冲刷已接受的消息然后关闭 stdin 并等待 2 秒PROCESS_TERMINATION_TIMEOUT若进程仍未退出POSIX 下先发SIGTERM、再等 2 秒FORCE_KILL_TIMEOUT后升级为SIGKILLWindows 下则直接通过 Job Object 强杀整个进程树。stderr 的默认去向与重定向子进程的 stderr默认流向你的 stderr。若想改道可以自己用stdio_client从mcp模块导入构造传输并传给它Client(stdio_client(server, errloglog_file))子进程不继承你的环境白名单机制警告子进程不会继承你的完整环境。它得到的是一份最小白名单——POSIX 下仅包含HOME、LOGNAME、PATH、SHELL、TERM、USERWindows 下是另一组见 src/mcp/client/stdio.py 的DEFAULT_INHERITED_ENV_VARS。这样做的目的是防止敏感信息泄漏到一个你未必编写的进程中。因此一个需要 API 密钥的服务器在子进程环境中找不到它。必须通过env显式传入这些变量会被合并merge在白名单之上——这正是上面示例中BOOKSHOP_API_KEY的作用。stdio_client内部正是用get_default_environment() | (server.env or {})合并出最终环境见 src/mcp/client/stdio.py。内存传输测试与嵌入式集成的首选在测试场景中既不需要部署什么也不需要启动什么——直接把服务器对象本身传给Client即可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)没有子进程、没有端口、线上没有任何字节。客户端和服务器只是同一进程中的两个对象——但调用依然走完整的真实协议层search_books会被正常地列出list、校验validate并调用invoke与通过 HTTP 调用时别无二致。docs/get-started/testing.md 的整个测试模式都建立在它之上。从源码看这一路径由 src/mcp/client/_memory.py 的InMemoryTransport实现它通过create_client_server_memory_streams()建立内存流并在后台任务中运行服务器。同样的形式还承担着嵌入式集成 API的职责一个自己构建服务器的应用可以在没有任何网络跳转的情况下直接调用其工具。SSE已被 Streamable HTTP 取代的旧传输sse_client(url)位于mcp.client.sse模块是被 Streamable HTTP 取代的旧 HTTP 传输。如果你需要与仍然只讲 SSE 的服务器通信可以用同样的方式包装它Client(sse_client(http://localhost:8000/sse))但不要在其之上构建任何新东西——新项目请直接使用 Streamable HTTP。Transport 协议所有传输的统一抽象对Client而言上文提到的所有传输本质上是同一件事。一个transport就是任何产生一对(read, write)消息流的异步上下文管理器——形式上即mcp.client中的Transport协议见 src/mcp/client/_transport.pyclass Transport(AbstractAsyncContextManager[TransportStreams], Protocol): Protocol for MCP transports. ...其中TransportStreams tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]]。Client按类型解析其参数str变成streamable_http_client(url)StdioServerParameters变成stdio_client(params)服务器对象则进程内直连而其他任何对象都直接作为 transport 打开。正是最后这条规则解释了为什么stdio_client(...)、streamable_http_client(...)、sse_client(...)都能插入同一个位置也解释了为什么你可以编写自己的传输——只要实现一个产出(read, write)流的异步上下文管理器即可例如以 WebSocket、Unix 域套接字等自定义通道承载 MCP 消息。快速回顾Client(http://.../mcp)一个 URL通过Streamable HTTP连接这是生产环境使用的传输。请求头、认证、代理、超时等全部配置在你传入streamable_http_client(url, http_client...)的httpx2.AsyncClient上不存在headers关键字。重定向只在 URL 自身的源内跟随如尾斜杠307/308重定向外加同一主机上的http→https。其他任何情况都会以Redirect to … not followed失败——请直接配置最终 URL。stdio 的写法是Client(StdioServerParameters(...))。只有当你想重定向子进程的 stderr 时才需要自己用stdio_client(...)包装。子进程获得的是白名单环境而不是你的环境env在其之上追加。Client(mcp)服务器对象走内存传输用于测试或把服务器嵌入到构建它的应用中。一个 transport 就是任何可以async with x as (read, write)的对象。Client会把既不是服务器对象、也不是 URL、也不是StdioServerParameters的任何东西直接交给这个协议。构造Client只是选择传输async with才真正打开它。一旦传输被打开通信双方还需要就协议版本达成一致。通常你永远不会需要关心这件事真到需要关心的那天请参阅 docs/protocol-versions.md。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 MCP Python SDK 客户端传输层完全指南：Streamable HTTP、stdio、内存传输与自定义 Transport
MCP Python SDK 客户端传输层完全指南Streamable HTTP、stdio、内存传输与自定义 Transport【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk导读本文围绕官方 Python SDKpython-sdkModel Context Protocol 的 Python 客户端/服务端实现中Client的传输层展开系统讲解 Streamable HTTP、stdio 子进程、内存in-process与旧版 SSE 四种传输方式的选择、配置与底层原理。读完本文你将掌握如何用一行 URL 连接生产服务器、如何通过自建httpx2.AsyncClient注入认证与代理、如何以StdioServerParameters拉起本地子进程以及为何所有传输最终都归结为同一个Transport协议。传输层是什么Client 如何根据类型推断连接方式每一个Client都是通过transport传输层与服务器对话的——传输层是真正承载消息的通道它决定了 JSON-RPC 报文以何种物理方式在客户端与服务器之间流动。在python-sdk中你永远不需要单独配置传输层。Client只接受一个位置参数server并完全根据该参数的类型推断出要使用的传输方式。这一点在源码中体现得极为直接Client的server字段类型被声明为Server[Any] | MCPServer | Transport | StdioServerParameters | str见 src/mcp/client/client.py而在__post_init__中按类型完成连接器的解析见 src/mcp/client/client.py传入Client(...)的参数类型实际使用的传输http://localhost:8000/mcpstrURLstreamable_http_client(url)StdioServerParameters(...)数据类stdio_client(params)以子进程方式运行Server(...)/MCPServer(...)对象服务器对象进程内in-process直接连接其他任意对象Transport实例直接作为传输层打开正因为这层“按类型分发”的设计stdio_client(...)、streamable_http_client(...)、sse_client(...)以及你自定义的传输都能无缝插入同一个位置。本篇文章的后续章节本质上是这张表的逐项展开。构造 ≠ 连接async with才是打开传输的时刻一个刚构造出来的Client并没有连接。构造过程只完成两件事按类型解析出传输层、准备好客户端会话真正的连接解析域名、建立 HTTP 连接、拉起子进程、完成协议握手全部发生在进入async with时。如果在进入上下文管理器之前就尝试访问连接例如访问client.sessionSDK 会明确报错RuntimeError: Client must be used within an async context manager该错误在 src/mcp/client/client.py 的session属性中抛出——_session只有在__aenter__握手成功之后才会被赋值。换句话说写下Client(http://...)这一行时什么都没发生没有解析域名、没有发起请求、没有启动子进程。这一行是“免费”的它只完成了传输层的选择。Streamable HTTP生产环境的首选传输将服务器端点 URL 以字符串形式传给Client即可获得Streamable 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])这就是完整的生产级客户端。Client内部将 URL 包装进streamable_http_client(...)其底层是基于一个按 MCP 需求配置的httpx2.AsyncClient连接connect/写入write/连接池pool超时均为 30 秒读取read超时为 300 秒——因为服务器可能长时间保持响应流打开如订阅、进度通知等场景。这组默认值定义在 src/mcp/shared/_httpx_utils.py 中并通过create_mcp_http_client()构建MCP_DEFAULT_TIMEOUT 30.0 # 一般操作秒 MCP_DEFAULT_SSE_READ_TIMEOUT 300.0 # SSE 流5 分钟秒从源码看src/mcp/client/streamable_http.pystreamable_http_client的签名是(url, *, http_clientNone, terminate_on_closeTrue)若未提供http_clientSDK 会调用create_mcp_http_client()创建一个带上述默认超时的客户端并在退出上下文时负责关闭它terminate_on_closeTrue默认值意味着退出上下文时若已建立会话传输层会向服务器发送一条DELETE 请求来终止会话见 src/mcp/client/streamable_http.py若服务器返回405说明它不支持会话终止SDK 会静默忽略。传输层还会为每次 HTTP 请求自动附加必要的 MCP 头accept: application/json, text/event-stream、content-type: application/json以及会话建立后的mcp-session-id与协商出的MCP-Protocol-Version见 src/mcp/client/streamable_http.py。自带 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])这段代码有两点需要特别注意客户端的所有权在你手中httpx2.AsyncClient是你创建的因此由你负责进入和退出它async with httpx2.AsyncClient(...)。SDK永远不会关闭一个不是它自己创建的客户端——上例中streamable_http_client只负责在传输层内部使用这个客户端其生命周期完全由外部async with管理。返回的是传输对象streamable_http_client(url, http_client...)返回一个 transportClient(transport)像接受任何其他传输一样接受它——这再次印证了上文的类型分发表。关于 TLS 的一个说明httpx2验证证书时使用的是操作系统信任库通过truststore实现而不是 SDK 内置的 CA 列表。如果运行环境没有可用的系统 CA 存储例如某些精简容器镜像有两种补救方式设置标准的SSL_CERT_FILE/SSL_CERT_DIR环境变量或为你的httpx2.AsyncClient显式传入verifyssl_context。相关背景可参阅 docs/migration.md。警惕旧习惯headers与timeout已从传输层移除警告streamable_http_client曾经直接接受headers和timeout参数现在不再支持。它的全部参数只有url、http_client和terminate_on_close。如果你按旧习惯写出streamable_http_client(url, headers...)会得到TypeError: streamable_http_client() got an unexpected keyword argument headers所有 HTTP 层面的配置请求头、超时、认证等现在都统一放在你传入的唯一一个httpx2.AsyncClient上。这也是 OAuth 接入的地方——httpx2.AsyncClient(authOAuthClientProvider(...))完整流程见 docs/client/oauth-clients.md。httpx2保留了httpx熟悉的 API如果你熟悉httpx就已经知道在这里如何做认证auth、代理proxy、事件钩子event hooks、重试retries和连接数限制connection limits。SDK 在 HTTP 层不增加也不削减任何东西——唯一的例外是下文的重定向处理。重定向策略只在同一源内跟随Streamable HTTP 传输只会连接到你给它的那个 URL且只连到该源origin。重定向的规则非常明确允许跟随保持相同 scheme、host、port 的307/308重定向同一主机上的http://→https://升级。这正好覆盖最常见的/mcp→/mcp/尾斜杠重定向。拒绝跟随指向任何其他位置的重定向一律不跟随调用会失败并报错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_redirects设置对 MCP 请求完全不被采纳正向、反向都是如此。SDK 自带的 OAuth 提供方对其自身请求也应用同样的规则。从源码看重定向判定逻辑在 src/mcp/shared/_httpx_utils.py 的_within_origin中实现比较(scheme, host, port)三元组是否完全一致或是否满足“同主机 http 默认端口 目标为 https”的升级条件。当重定向未被跟随时src/mcp/client/streamable_http.py 的_unfollowed_redirect会生成上述错误信息。实用提示如果遇到如下错误——Redirect to http://… not followed: it would downgrade this HTTPS endpoint to plain HTTP——说明服务器位于一个它自己不知道的TLS 终止代理之后且正在发出http://重定向。修复方式是在服务器端修正代理配置见 docs/run/deploy.md或者直接使用错误信息中建议的那个精确https://…/URL。stdio以子进程方式运行服务器stdio服务器本质上是一个子进程客户端负责启动它向它的 stdin 写入 JSON-RPC从它的 stdout 读取 JSON-RPC。这正是桌面主机desktop host在你机器上运行服务器的方式——一个主机本质上就是这段代码加一个界面。docs/get-started/real-host.md 从主机的视角展示了同样的关系以配置文件的形式呈现。用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])StdioServerParameters的完整字段定义在 src/mcp/client/stdio.py除示例中用到的command、args、env外还支持字段类型默认值说明commandstr必填启动服务器的可执行文件argslist[str][]传给可执行文件的命令行参数envdict[str, str] \| NoneNone额外环境变量合并在环境白名单之上cwdstr \| Path \| NoneNone启动进程时的工作目录encodingstrutf-8与服务器通信的文本编码encoding_error_handlerstrict \| ignore \| replacestrict编码错误处理方式进入async with块会启动子进程退出时会自动关闭子进程关闭 stdin、等待进程自行退出、若迟迟不退则强制终止。你永远不需要手动清理。从 src/mcp/client/stdio.py 的实现看关闭流程是先关闭读写流、给写者最多 0.5 秒冲刷已接受的消息然后关闭 stdin 并等待 2 秒PROCESS_TERMINATION_TIMEOUT若进程仍未退出POSIX 下先发SIGTERM、再等 2 秒FORCE_KILL_TIMEOUT后升级为SIGKILLWindows 下则直接通过 Job Object 强杀整个进程树。stderr 的默认去向与重定向子进程的 stderr默认流向你的 stderr。若想改道可以自己用stdio_client从mcp模块导入构造传输并传给它Client(stdio_client(server, errloglog_file))子进程不继承你的环境白名单机制警告子进程不会继承你的完整环境。它得到的是一份最小白名单——POSIX 下仅包含HOME、LOGNAME、PATH、SHELL、TERM、USERWindows 下是另一组见 src/mcp/client/stdio.py 的DEFAULT_INHERITED_ENV_VARS。这样做的目的是防止敏感信息泄漏到一个你未必编写的进程中。因此一个需要 API 密钥的服务器在子进程环境中找不到它。必须通过env显式传入这些变量会被合并merge在白名单之上——这正是上面示例中BOOKSHOP_API_KEY的作用。stdio_client内部正是用get_default_environment() | (server.env or {})合并出最终环境见 src/mcp/client/stdio.py。内存传输测试与嵌入式集成的首选在测试场景中既不需要部署什么也不需要启动什么——直接把服务器对象本身传给Client即可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)没有子进程、没有端口、线上没有任何字节。客户端和服务器只是同一进程中的两个对象——但调用依然走完整的真实协议层search_books会被正常地列出list、校验validate并调用invoke与通过 HTTP 调用时别无二致。docs/get-started/testing.md 的整个测试模式都建立在它之上。从源码看这一路径由 src/mcp/client/_memory.py 的InMemoryTransport实现它通过create_client_server_memory_streams()建立内存流并在后台任务中运行服务器。同样的形式还承担着嵌入式集成 API的职责一个自己构建服务器的应用可以在没有任何网络跳转的情况下直接调用其工具。SSE已被 Streamable HTTP 取代的旧传输sse_client(url)位于mcp.client.sse模块是被 Streamable HTTP 取代的旧 HTTP 传输。如果你需要与仍然只讲 SSE 的服务器通信可以用同样的方式包装它Client(sse_client(http://localhost:8000/sse))但不要在其之上构建任何新东西——新项目请直接使用 Streamable HTTP。Transport 协议所有传输的统一抽象对Client而言上文提到的所有传输本质上是同一件事。一个transport就是任何产生一对(read, write)消息流的异步上下文管理器——形式上即mcp.client中的Transport协议见 src/mcp/client/_transport.pyclass Transport(AbstractAsyncContextManager[TransportStreams], Protocol): Protocol for MCP transports. ...其中TransportStreams tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]]。Client按类型解析其参数str变成streamable_http_client(url)StdioServerParameters变成stdio_client(params)服务器对象则进程内直连而其他任何对象都直接作为 transport 打开。正是最后这条规则解释了为什么stdio_client(...)、streamable_http_client(...)、sse_client(...)都能插入同一个位置也解释了为什么你可以编写自己的传输——只要实现一个产出(read, write)流的异步上下文管理器即可例如以 WebSocket、Unix 域套接字等自定义通道承载 MCP 消息。快速回顾Client(http://.../mcp)一个 URL通过Streamable HTTP连接这是生产环境使用的传输。请求头、认证、代理、超时等全部配置在你传入streamable_http_client(url, http_client...)的httpx2.AsyncClient上不存在headers关键字。重定向只在 URL 自身的源内跟随如尾斜杠307/308重定向外加同一主机上的http→https。其他任何情况都会以Redirect to … not followed失败——请直接配置最终 URL。stdio 的写法是Client(StdioServerParameters(...))。只有当你想重定向子进程的 stderr 时才需要自己用stdio_client(...)包装。子进程获得的是白名单环境而不是你的环境env在其之上追加。Client(mcp)服务器对象走内存传输用于测试或把服务器嵌入到构建它的应用中。一个 transport 就是任何可以async with x as (read, write)的对象。Client会把既不是服务器对象、也不是 URL、也不是StdioServerParameters的任何东西直接交给这个协议。构造Client只是选择传输async with才真正打开它。一旦传输被打开通信双方还需要就协议版本达成一致。通常你永远不会需要关心这件事真到需要关心的那天请参阅 docs/protocol-versions.md。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考