python-sdk 的 MCP 客户端 `Client`:连接、生命周期与全部协议操作实战指南

发布时间:2026/9/21 16:04:07

python-sdk 的 MCP 客户端 `Client`:连接、生命周期与全部协议操作实战指南 python-sdk 的 MCP 客户端Client连接、生命周期与全部协议操作实战指南【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk本篇指南以 Model Context ProtocolMCP官方 Python SDKpython-sdk的Client为核心讲解 Python 程序如何通过一个对象、一个生命周期与 MCP 服务器通信从 URL、子进程、自定义传输到进程内连接四种接入方式再到工具列表与调用、资源读写、提示词渲染、自动补全与分页的完整协议操作并辅以源码级实现佐证。读完你可以独立编写出连接任意 MCP 服务器的客户端程序并理解其底层会话、类型结果与错误语义。概述一个对象、一个生命周期Client是 Python 程序与 MCP 服务器通信的手段。它遵循一个对象、一个生命周期的设计构造它进入async with调用方法。所有协议动词列出工具、调用工具、读取资源、渲染提示词都是返回类型化结果的async方法。没有connect()/close()配对——进入async with即连接并协商退出即断开块结束后Client不可复用。在源码中Client定义于 src/mcp/client/client.py其__aenter__src/mcp/client/client.py#L447完成连接的建立与协议协商__aexit__通过AsyncExitStack统一拆除传输层。值得注意的是会话只有在握手成功之后才对外发布self._session session因此只要进入块内protocol_version与server_capabilities一定已填充完毕。第一个客户端Bookshop 示例客户端需要与之对话的服务器。本页所有示例连接的服务器都是这个 Bookshop 服务器其完整定义见 docs_src/client/tutorial001.py将它保存为server.py并通过 HTTP 运行from pydantic import BaseModel from mcp.server import MCPServer from mcp.server.mcpserver.exceptions import ToolError from mcp.types import Completion, CompletionArgument, CompletionContext, PromptReference, ResourceTemplateReference mcp MCPServer(Bookshop, instructionsSearch the catalog before recommending a book.) GENRES [fiction, non-fiction, poetry] class Book(BaseModel): title: str author: str year: int mcp.tool(titleSearch the catalog) def search_books(query: str, limit: int 10) - str: Search the catalog by title or author. return fFound 3 books matching {query!r} (showing up to {limit}). mcp.tool() def lookup_book(title: str) - Book: Look up a book by its exact title. if title ! Dune: raise ToolError(fNo book titled {title!r} in the catalog.) return Book(titleDune, authorFrank Herbert, year1965) mcp.resource(catalog://genres) def genres() - list[str]: The genres the catalog is organised by. return GENRES mcp.resource(catalog://genres/{genre}) def books_in_genre(genre: str) - str: Every title we stock in one genre. return f3 books filed under {genre}. mcp.prompt(titleRecommend a book) def recommend(genre: str) - str: Ask for a recommendation in a genre. return fRecommend one {genre} book from the catalog and say why. mcp.completion() async def complete_genre( ref: PromptReference | ResourceTemplateReference, argument: CompletionArgument, context: CompletionContext | None, ) - Completion | None: return Completion(values[genre for genre in GENRES if genre.startswith(argument.value)])在第一个终端启动服务器uv run mcp run server.py --transport streamable-http这将在http://localhost:8000/mcp提供服务。客户端是独立的程序将下面的代码保存为client.py在第二个终端运行python client.pyimport anyio from mcp import Client async def main() - None: async with Client(http://localhost:8000/mcp) as client: print(client.server_info) print(client.server_capabilities) print(client.protocol_version) print(client.instructions) if __name__ __main__: anyio.run(main)关键点Client(http://localhost:8000/mcp)传入的是URL因此通过 Streamable HTTP 连接到刚启动的服务器async with就是生命周期。进入时连接并协商退出时断开。没有connect()/close()配对块结束后不能复用该Client块内部连接事实已经以普通属性property的形式就绪。这个示例的运行入口使用anyio.run(main)因为 SDK 底层基于 anyio 抽象可无缝切换 asyncio/trio 后端。可以传给Client的四种对象Client只接受一个位置参数并根据其类型解析传输方式对应 src/mcp/client/client.py#L283 的类型注解与 src/mcp/client/client.py#L388 的__post_init__分支传入对象连接方式典型场景URL 字符串Client(http://localhost:8000/mcp)Streamable HTTP即生产部署常用的传输连接远程或本机 HTTP 服务StdioServerParameters将命令作为本地子进程启动通过其 stdin/stdout 通信本地 CLI 式 MCP 服务器任意传输Transport凡是能async with ... as (read, write)的对象都行自研传输、自定义 HTTP 客户端MCPServer或低层Server实例进程内直接连接无子进程、无端口测试场景URLStreamable HTTPURL 字符串会被包装进streamable_http_clientsrc/mcp/client/streamable_http.py#L680。该传输函数接受可选的http_client预配置的httpx2.AsyncClient可携带自定义 headers、认证等与terminate_on_close退出时是否发送 DELETE 终止会话默认True。若未提供http_clientSDK 会以推荐 MCP 超时创建默认客户端。StdioServerParameters子进程StdioServerParameterssrc/mcp/client/stdio.py#L94是 Pydantic 模型字段包括command要运行的可执行文件必填args命令行参数列表env额外环境变量与默认环境合并cwd启动进程的工作目录encoding消息文本编码默认utf-8encoding_error_handler编码错误处理方式strict/ignore/replace默认strict。stdio_client(server)src/mcp/client/stdio.py#L114负责派生子进程并通过 stdin/stdout 建链。自定义传输任何实现了传输协议可async with ... as (read, write)的对象都可以直接传入例如围绕自研 HTTP 客户端包装的streamable_http_client(url, http_client...)。这与 Client 构造函数中非 URL、非StdioServerParameters、非Server一律按传输处理的兜底分支src/mcp/client/client.py#L397一致。头部、子进程细节、超时与Transport协议在单独的文档 客户端传输 中阐述。MCPServer进程内测试MCPServer或低层Server实例会被__post_init__识别src/mcp/client/client.py#L389通过内存传输直接建链无子进程、无端口。这是为测试设计的测试 一节建立在其之上。本页其余内容对四种方式完全一致。额外的构造参数raise_exceptions、read_timeout_seconds、sampling_callback、list_roots_callback、logging_callback、log_level、message_handler、client_info、mode、prior_discover、elicitation_callback、input_required_max_rounds、extensions、cache等详见 src/mcp/client/client.py#L294-L366可深入源码查阅。已连接客户端上的四个属性进入块的那一刻以下四个只读属性即被填充client.server_info服务器身份信息。若 2026 时代的服务器不报告身份则为Nonepython-sdk 服务器默认报告。此处server_info.name为Bookshopserver_info.version为服务器上报的版本client.server_capabilities服务器能做什么tools、resources、prompts、completions……。服务器不具备的能力对应Noneclient.protocol_version双方协商一致的协议版本。此处为2026-07-28client.instructions服务器的instructions字符串未设置则为None。你从未手动选择协议版本。默认情况下Client会探测服务器modeauto在旧服务器上回退到经典握手initialize因此一个客户端可以对接任何时代的服务器。需要控制协商时参见 协议版本。协商模式在源码中由mode: ConnectMode auto控制src/mcp/client/client.py#L330__aenter__按legacy强制初始化握手、auto探测 回退、或直接采纳现代版本session.adopt(...)三条路径分别处理src/mcp/client/client.py#L457-L462。提示client.session是底层ClientSession属于低层逃生舱escape hatch本页内容用不到它。列出工具完整示例见 docs_src/client/tutorial002.pyimport anyio from mcp import Client async def main() - None: async with Client(http://localhost:8000/mcp) as client: result await client.list_tools() for tool in result.tools: print(tool.name) print(tool.title) print(tool.description) print(tool.input_schema) if __name__ __main__: anyio.run(main)list_tools()返回ListToolsResult工具位于.tools。每个工具都是宿主host要交给模型的完整定义。第一个工具如下tool.name # search_books tool.title # Search the catalog tool.description # Search the catalog by title or author.tool.input_schema是服务器根据函数类型提示推导出的 JSON Schema{ type: object, properties: { query: {title: Query, type: string}, limit: {default: 10, title: Limit, type: integer} }, required: [query], title: search_booksArguments }这个 Schema 既是 UI 渲染参数表单所需的全部信息也是模型生成合法参数所需的全部信息。第二个工具lookup_book注册时没有传title所以它的tool.title是None。提示title是可选的因此向人类展示工具的 UI 需要抉择有title用title没有则用name。from mcp.shared.metadata_utils import get_display_name恰好做这件事src/mcp/shared/metadata_utils.py#L10且适用于工具、资源、资源模板和提示词。从源码可见其优先级规则工具为title annotations.title name其余对象为title namesrc/mcp/shared/metadata_utils.py#L35-L45。底层实现上list_tools()src/mcp/client/client.py#L924会经由_cached_fetch走tools/list方法并支持响应缓存cache_mode列表命中缓存时还会让会话重新吸收工具列表以重建派生的按工具状态。调用工具call_tool(name, arguments)执行工具并返回CallToolResult。完整示例见 docs_src/client/tutorial003.pyimport anyio from mcp import Client from mcp.types import TextContent async def main() - None: async with Client(http://localhost:8000/mcp) as client: result await client.call_tool(lookup_book, {title: Dune}) for block in result.content: if isinstance(block, TextContent): print(block.text) print(result.structured_content) print(result.is_error) if __name__ __main__: anyio.run(main)服务器的lookup_book返回 PydanticBook。客户端看到的是result.content # [TextContent(typetext, text{\n title: Dune,\n author: Frank Herbert,\n year: 1965\n})] result.structured_content # {title: Dune, author: Frank Herbert, year: 1965} result.is_error # False一个返回值三样可读的东西各自有不同的消费者CallToolResult类型定义见 src/mcp-types/mcp_types/_types.py#L1463。content模型读的部分content是内容块的list而内容块是联合类型TextContent、ImageContent、AudioContent、ResourceLink或EmbeddedResource。一个工具可以返回多个不同类型的块。这正是main在触碰block.text之前用isinstance(block, TextContent)收窄类型的原因。注意isinstance之外没有.text类型检查器不允许因为ImageContent拥有的是.data而不是.text。联合类型诚实地暴露了工具允许发送的内容你的代码也应如此。structured_content应用读的部分structured_content是工具返回值的 JSON 表达与工具声明的output_schema一致。不需要字符串解析不需要猜测。两者同时存在时是刻意同一件事说两遍content给模型structured_content给代码。结构化部分从何而来、如何控制参见 结构化输出 页面。is_error工具是否失败抛出异常的工具不会在客户端抛异常而是作为一个普通的is_errorTrue结果返回。验证向lookup_book请求Solaris目录中不存在的书名函数抛出ToolError定义于 src/mcp/server/mcpserver/exceptions.py。但调用仍正常返回result.is_error # True result.content # [TextContent(typetext, textError executing tool lookup_book: No book titled Solaris in the catalog.)] result.structured_content # NoneToolError的消息落在了content里模型可以读到并重试。这是刻意的设计工具错误是对话的一部分而不是崩溃。若工具因其他异常崩溃content只会写Error executing tool lookup_book。在信任structured_content之前永远先看is_error。警告is_errorTrue覆盖的范围比你自己raise更广。请求一个服务器根本没有的工具call_tool(does_not_exist, {})同样不会抛异常返回同样的形状is_errorTruecontent中含Unknown tool: does_not_exist。Client方法只会在服务器以 JSON-RPC错误而非结果应答时抛出MCPError。服务器何时产生哪种应答见 错误处理。这与类型定义中的语义一致源自工具的差错应在结果中以is_errortrue报告好让 LLM 看到并自我修正而不是 MCP 协议级错误src/mcp-types/mcp_types/_types.py#L1466-L1469。从实现看call_tool()src/mcp/client/client.py#L751还支持read_timeout_seconds单轮超时、progress_callback进度回调、meta等参数若服务器返回InputRequiredResult客户端会自动把嵌入的输入请求分发给 sampling / elicitation / roots 回调并重试上限为input_required_max_rounds。此外它还支持is_errorFalse时的输出 Schema 再校验validate_tool_result。资源资源动词成对出现两种列举方式一种读取方式。完整示例见 docs_src/client/tutorial004.pyimport anyio from mcp import Client from mcp.types import TextResourceContents async def main() - None: async with Client(http://localhost:8000/mcp) as client: listed await client.list_resources() print([resource.uri for resource in listed.resources]) templates await client.list_resource_templates() print([template.uri_template for template in templates.resource_templates]) result await client.read_resource(catalog://genres/poetry) for contents in result.contents: if isinstance(contents, TextResourceContents): print(contents.text) if __name__ __main__: anyio.run(main)list_resources()返回具体的资源即 URI 固定的资源。此处为[catalog://genres]list_resource_templates()返回参数化的资源。此处为[catalog://genres/{genre}]。两者是两份不同的列表因为模板在填入值之前不可读read_resource(uri)接受普通strURI对两者都有效传入catalog://genres/poetry服务器将其匹配到模板。read_resource返回contents是TextResourceContents或BlobResourceContents的列表。与工具内容同理先用isinstance收窄再读.text或.blob。实现层面src/mcp/client/client.py#L592-L686三个方法都走resources/list、resources/templates/list、resources/read底层会话方法支持cursor与cache_moderead_resource同样具备InputRequiredResult自动驱动与缓存失效逻辑带meta的调用永远直达服务器。客户端还可以获知资源何时变更。2025 时代的连接上使用subscribe_resource(uri)/unsubscribe_resource(uri)——这是MCPServer不实现的方法对因此在 2026-07-28 线路上这些动词已不存在请求会以-32601、Method not found应答。2026 年的替代方案是subscriptions/listen流MCPServer确实提供此处server_capabilities.resources.subscribe为True。在源码中这对 2025 时代方法已被标记为deprecated并明确提示resources/subscribe 已随 2026-07-28 移除请改用Client.listen()src/mcp/client/client.py#L735-L749。用client.listen(...)消费该流的完整方式见本节的 订阅 页面。listen()的签名支持tools_list_changed、prompts_list_changed、resources_list_changed与resource_subscriptions四个过滤参数src/mcp/client/client.py#L688。提示词完整示例见 docs_src/client/tutorial005.pyimport anyio from mcp import Client async def main() - None: async with Client(http://localhost:8000/mcp) as client: listed await client.list_prompts() print(listed.prompts) result await client.get_prompt(recommend, {genre: poetry}) for message in result.messages: print(message.role, message.content) if __name__ __main__: anyio.run(main)list_prompts()告诉你服务器提供什么、每个提示词需要什么prompt.name # recommend prompt.title # Recommend a book prompt.arguments # [PromptArgument(namegenre, requiredTrue)]get_prompt(name, arguments)渲染提示词。参数字典是str - str提示词参数永远是字符串。结果是messages即PromptMessage的列表每个消息都有role和content块message.role # user message.content # TextContent(typetext, textRecommend one poetry book from the catalog and say why.)宿主把这些消息直接交给模型。这个功能就这么简单——但注意get_promptsrc/mcp/client/client.py#L842同样支持input_responses/request_state与InputRequiredResult自动驱动。自动补全带补全处理器的服务器可以在用户输入过程中自动补全提示词和资源模板参数。完整示例见 docs_src/client/tutorial006.pyimport anyio from mcp import Client from mcp.types import PromptReference async def main() - None: async with Client(http://localhost:8000/mcp) as client: result await client.complete( refPromptReference(typeref/prompt, namerecommend), argument{name: genre, value: p}, ) print(result.completion.values) if __name__ __main__: anyio.run(main)ref指明你在填充哪个提示词或模板PromptReference或ResourceTemplateReferenceargument是{name: ..., value: ...}参数以及用户到目前为止输入的值。答案在result.completion.values。输入p服务器返回[poetry]。服务端实现、以及处理器如何利用其他已填充参数收窄建议见 自动补全 页面。complete()方法src/mcp/client/client.py#L906还接受可选的context_arguments用于提供额外上下文参数。分页每个list_*方法都接受cursor关键字每个结果都携带next_cursor。当next_cursor为None时你已经拿全了。完整示例见 docs_src/client/tutorial007.pyimport anyio from mcp import Client from mcp.types import Tool async def list_all_tools(client: Client) - list[Tool]: tools: list[Tool] [] cursor: str | None None while True: page await client.list_tools(cursorcursor) tools.extend(page.tools) if page.next_cursor is None: return tools cursor page.next_cursor async def main() - None: async with Client(http://localhost:8000/mcp) as client: tools await list_all_tools(client) print([tool.name for tool in tools]) if __name__ __main__: anyio.run(main)list_all_tools对任何服务器都是正确的。MCPServer把所有内容放在一页返回因此next_cursor为None、循环只执行一次——这也是大多数代码从不写这个循环的原因。真正分页的服务器与游标遵循的规则见 分页。在测试中使用本页每个client.py都是通过 HTTP 连接server.py的。在测试中你可以跳过网络直接把服务器对象交给Clientfrom server import mcp然后Client(mcp)。没有进程、没有端口上面所有方法行为一致。为此专门设计了一个构造标志Client(mcp, raise_exceptionsTrue)。它只对进程内连接生效在 src/mcp/client/client.py#L295 中定义默认为False底层通过InMemoryTransport与直接分发器对create_direct_dispatcher_pair(raise_handler_exceptionsraise_exceptions)实现src/mcp/client/client.py#L108-L114。raise_exceptionsTrue使服务器端未映射的处理错误直接抛出False则对其进行脱敏处理——仓库测试 tests/client/test_client.py#L218-L244 对两条路径都有覆盖验证。完整的模式讲解见 测试 页面。小结Client(x)URL 字符串走 Streamable HTTPStdioServerParameters启动子进程传输对象直接进入测试中则接收服务器对象本身async with就是整个生命周期。块内server_capabilities与protocol_version已填充服务器提供时server_info与instructions亦然list_tools()给出每个工具的name、title、description与input_schemacall_tool()返回给模型的content、给代码的structured_content以及is_error。抛异常的工具是结果而非异常content是块类型的联合读取前先用isinstance收窄list_resources/list_resource_templates/read_resource、list_prompts/get_prompt、complete构成其余动词每个list_*都接受cursor循环直到next_cursor为None。服务器可以向客户端请求什么、如何应答见 客户端回调。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
延伸阅读

更多相关文章

2026/9/21 16:04:07

[Question]

人工智能大模型AI Agent自主智能体Agent 编排MCP Clients知识图谱 【免费下载链接】openfang Open-source Agent Operating System 项目地址: https://gitcode.com/gh_mirrors/op/openfang 点击查看 免费下载 Date: YYYY-MM-DD | Sources: N | Confidence: high/me…

2026/9/21 16:04:07

vscode 无法登录 GitHub Coplilot 插件,点击登录链接没有反应

1.问题描述 vscode 无法登录使用 GitHub Coplilot 插件,点击 sign in wirh a github account 没反应,点击企业登录可以,卸载重装插件不管用。 2.解决 修改配置文件即可。 ctrl shift p ,输入settings,点击用户设…

2026/9/21 3:28:31

GAMP 5 基于风险的计算机化系统验证:软件分类与审计追踪实践

简介:《A Risk-Based Approach to Compliant GxP Computerized Systems》即业内熟知的GAMP 5指南,面向制药企业质量与IT合规人员、验证工程师及计算机化系统管理者,用于解决GxP法规环境下系统合规性难以科学落地的问题。文档以风险管理为主线…

2026/9/21 3:33:19

安全托管MSSP实战:从静态防御到人机协同的攻防运营与应急响应

简介:这份PPT围绕互联网业务安全托管服务展开,面向企业安全负责人、IT运维人员及关注MSSP/MSS选型的读者,重点回应传统安全过度依赖人工、碎片化静态防御难以对抗产业化攻击等痛点。资源共1个pptx文件,包体约30.63MB,以…

2026/9/21 0:02:23

OpenResearch:构建可复现的开放式研究工作流

第一次看到“OpenResearch”这个名字,我脑子里冒出的不是某个具体软件,而更像一种研究方式的宣言:开放、可复现、可验证。这三件事放在一起,其实比大多数人想象中难得多。过去几年我一直在折腾自己的研究工作流,从纯纸…

2026/9/20 4:54:47

USB Type-C PCB布局分区设计:电源、高速信号与PD协议全攻略

做硬件这行,Type-C接口算是典型的“看着简单,做起来全坑”的东西。光引脚就24个,高低速信号、电源、控制线全部塞在一个小小的连接器里,如果PCB布局不做规划,打样回来基本就是“插上没反应”、“高速掉线”、“静电一打…

2026/9/20 5:01:23

系统编程学习原型如何补齐稳定性边界

系统编程学习原型如何补齐稳定性边界预算有限时&#xff0c;我先优化明显多余的复制&#xff0c;而不是猜测性地换容器。用借用传递只读数据通常就能减少分配&#xff1a; fn parse(line: &str) -> Result<Item, Error> { /* ... */ }用基准确认热点确实在分配&am…

2026/9/21 10:29:02

雨花区哪家财务公司代理记账比较好?

在雨花区&#xff0c;企业处理财税事务常常面临诸多挑战&#xff0c;选择一家靠谱的财务公司至关重要。湖南巨勤财务管理咨询有限公司就是本地正规实体财税服务机构&#xff0c;深耕本地工商财税行业多年&#xff0c;熟悉当地工商局、税务局最新政策与申报流程。主营公司注册、…

还想了解更多?直接咨询顾问

免费诊断 + 免费方案 + 透明报价。

全国咨询热线400-8866-253
免费获取方案
咨询二维码