企业官网建设流程全解析

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 Protocol（MCP）官方 Python SDK（python-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", instructions="Search the catalog before recommending a book.") GENRES = ["fiction", "non-fiction", "poetry"] class Book(BaseModel): title: str author: str year: int @mcp.tool(title="Search the catalog") def search_books(query: str, limit: int = 10) -> str: """Search the catalog by title or author.""" return f"Found 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(f"No book titled {title!r} in the catalog.") return Book(title="Dune", author="Frank Herbert", year=1965) @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 f"3 books filed under {genre}." @mcp.prompt(title="Recommend a book") def recommend(genre: str) -> str: """Ask for a recommendation in a genre.""" return f"Recommend 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.py：

import 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：Streamable HTTP

URL 字符串会被包装进streamable_http_client（src/mcp/client/streamable_http.py#L680）。该传输函数接受可选的http_client=（预配置的httpx2.AsyncClient，可携带自定义 headers、认证等）与terminate_on_close（退出时是否发送 DELETE 终止会话，默认True）。若未提供http_client，SDK 会以推荐 MCP 超时创建默认客户端。

StdioServerParameters：子进程

StdioServerParameters（src/mcp/client/stdio.py#L94）是 Pydantic 模型，字段包括：

• command：要运行的可执行文件（必填）；
• args：命令行参数列表；
• env：额外环境变量，与默认环境合并；
• cwd：启动进程的工作目录；
• encoding：消息文本编码，默认"utf-8"；
• encoding_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 时代的服务器不报告身份则为None（python-sdk 服务器默认报告）。此处server_info.name为"Bookshop"，server_info.version为服务器上报的版本；
• client.server_capabilities：服务器能做什么（tools、resources、prompts、completions……）。服务器不具备的能力对应None；
• client.protocol_version：双方协商一致的协议版本。此处为"2026-07-28"；
• client.instructions：服务器的instructions=字符串，未设置则为None。

你从未手动选择协议版本。默认情况下Client会探测服务器（mode="auto"），在旧服务器上回退到经典握手（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.py：

import 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 > name（src/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.py：

import 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(type='text', 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_error=True结果返回。

验证：向lookup_book请求"Solaris"（目录中不存在的书名），函数抛出ToolError（定义于 src/mcp/server/mcpserver/exceptions.py）。但调用仍正常返回：

result.is_error # True result.content # [TextContent(type='text', text="Error executing tool lookup_book: No book titled 'Solaris' in the catalog.")] result.structured_content # None

ToolError的消息落在了content里，模型可以读到并重试。这是刻意的设计：工具错误是对话的一部分，而不是崩溃。（若工具因其他异常崩溃，content只会写Error executing tool lookup_book。）在信任structured_content之前，永远先看is_error。

警告：is_error=True覆盖的范围比你自己raise更广。请求一个服务器根本没有的工具（call_tool("does_not_exist", {})），同样不会抛异常，返回同样的形状：is_error=True，content中含Unknown tool: does_not_exist。Client方法只会在服务器以 JSON-RPC错误（而非结果）应答时抛出MCPError。服务器何时产生哪种应答，见 错误处理。这与类型定义中的语义一致：源自工具的差错应在结果中以is_error=true报告（好让 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_error=False时的输出 Schema 再校验（validate_tool_result）。

资源

资源动词成对出现：两种列举方式，一种读取方式。完整示例见 docs_src/client/tutorial004.py：

import 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_mode，read_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.py：

import 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(name='genre', required=True)]

get_prompt(name, arguments)渲染提示词。参数字典是str -> str：提示词参数永远是字符串。结果是messages，即PromptMessage的列表，每个消息都有role和content块：

message.role # 'user' message.content # TextContent(type='text', text='Recommend one poetry book from the catalog and say why.')

宿主把这些消息直接交给模型。这个功能就这么简单——但注意get_prompt（src/mcp/client/client.py#L842）同样支持input_responses/request_state与InputRequiredResult自动驱动。

自动补全

带补全处理器的服务器可以在用户输入过程中自动补全提示词和资源模板参数。完整示例见 docs_src/client/tutorial006.py：

import 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( ref=PromptReference(type="ref/prompt", name="recommend"), argument={"name": "genre", "value": "p"}, ) print(result.completion.values) if __name__ == "__main__": anyio.run(main)

• ref指明你在填充哪个提示词或模板：PromptReference或ResourceTemplateReference；
• argument是{"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.py：

import 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(cursor=cursor) 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的。在测试中你可以跳过网络，直接把服务器对象交给Client：from server import mcp，然后Client(mcp)。没有进程、没有端口，上面所有方法行为一致。

为此专门设计了一个构造标志：Client(mcp, raise_exceptions=True)。它只对进程内连接生效（在 src/mcp/client/client.py#L295 中定义，默认为False），底层通过InMemoryTransport与直接分发器对（create_direct_dispatcher_pair(raise_handler_exceptions=raise_exceptions)）实现（src/mcp/client/client.py#L108-L114）。raise_exceptions=True使服务器端未映射的处理错误直接抛出，False则对其进行脱敏处理——仓库测试 tests/client/test_client.py#L218-L244 对两条路径都有覆盖验证。完整的模式讲解见 测试 页面。

小结

• Client(x)：URL 字符串走 Streamable HTTP，StdioServerParameters启动子进程，传输对象直接进入，测试中则接收服务器对象本身；
• async with就是整个生命周期。块内server_capabilities与protocol_version已填充；服务器提供时server_info与instructions亦然；
• list_tools()给出每个工具的name、title、description与input_schema；
• call_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