企业官网建设流程全解析

python-sdk 客户端开发实战：用 PythonClient全面掌握 MCP 协议的每个动词

【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk

导读

本文围绕 python-sdk（Model Context Protocol 官方 Python SDK）中面向客户端编程的核心对象——Client展开。你将学会如何用一条 URL、一个本地子进程参数、一个自定义 transport 乃至一个内存中的服务器对象启动客户端，并完整掌握list_tools、call_tool、read_resource、get_prompt、complete等协议动词的调用方式与返回值语义，同时理解协议版本协商、分页、资源订阅与测试方法。文中所有示例均来自仓库 docs_src/client 目录下的可运行教程，并辅以 客户端源码 与测试进行源码级印证。

Client是什么：一个对象，一个生命周期

在 python-sdk 中，Client是 Python 程序与 MCP 服务器对话的入口。它的设计哲学是"一个对象、一个生命周期"：

• 构造：传入连接目标；
• 进入async with：连接并完成握手（negotiation）；
• 调用方法：协议中的每个动词——列出工具、调用工具、读取资源、渲染 prompt——都是该对象上的一个async方法，返回类型化的结果对象。

离开async with块即断开连接。没有connect()/close()这样的成对调用，而且一个Client在块结束后不能被复用。从源码结构看，Client类定义于 src/mcp/client/client.py#L262，其底层由ClientSession（见 src/mcp/client/session.py#L388）承载全部协议交互。

你的第一个客户端

客户端需要服务器配合才能演示。本文档页所有示例连接的都是同一个Bookshop服务器。先把它保存为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)])

这段完整代码位于 docs_src/client/tutorial001.py。它同时注册了工具、资源、prompt 与 completion 处理器，是本文所有客户端示例的对端。启动命令：

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)

这段代码来自 docs_src/client/tutorial001_client.py。三个要点：

• Client("http://localhost:8000/mcp")接收的是一个URL，因此通过Streamable HTTP传输连接到刚启动的服务器；
• async with就是生命周期：进入时连接并协商协议，退出时断开；
• 进入块之后，连接事实已经以普通属性（plain properties）的形式就绪，直接读取即可。

可以传给Client的四种参数

Client只接收一个位置参数，并根据其类型自动解析出对应的传输方式：

除传输方式外，本页其余内容对四种情况完全一致。关于自定义 header、子进程参数、超时与Transport协议本身的细节，见 客户端传输。其中进程内模式是测试的基石，测试 一节专门围绕它构建。

已连接客户端上有什么

进入async with块后，四个只读属性即被填充：

• 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会**探测（probe）**服务器，对旧代际服务器回退到经典握手，因此同一个客户端可以对接任何代际的服务器。需要精细控制时，详见 协议版本。

提示：client.session是底层的ClientSession，属于低层逃逸舱口，本页内容完全用不到它。

列出工具：list_tools()

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)

代码位于 docs_src/client/tutorial002.py。list_tools()返回一个ListToolsResult，工具位于.tools中。每个Tool都是宿主（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 既是界面渲染参数表单的全部依据，也是模型生成合法参数的全部依据。注意第二个工具lookup_book注册时没有传title=，因此其tool.title为None。

提示：title是可选字段，因此面向人类展示工具的界面需要自行取舍：有title用title，没有则用name。from mcp.shared.metadata_utils import get_display_name正好做了这件事，且同时适用于工具、资源、资源模板和 prompts。

调用工具：call_tool(name, arguments)

call_tool(name, arguments)执行工具并返回CallToolResult：

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)

代码位于 docs_src/client/tutorial003.py。服务器端的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

一个返回值，三处可读，各有各的消费方。从 client.py 源码 的call_tool实现可以看出，它会把服务器应答装配成包含上述三个字段的结果对象。

content：模型读的部分

content是一个内容块（content block）列表，而内容块是联合类型：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，但调用仍正常返回：

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-RPCerror（而非 result）应答时才抛出MCPError。服务器何时产生哪一种，见 处理错误。

资源：列出与读取

资源动词成对出现：两种列出方式、一种读取方式。

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)

代码位于 docs_src/client/tutorial004.py。三个要点：

• 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）。对应源码见 client.py#L624 的read_resource。

资源变更通知与订阅

客户端还可以被通知资源发生变化。在 2025 代际连接上，这是subscribe_resource(uri)/unsubscribe_resource(uri)这对方法——但MCPServer并不实现它们，因此在 2026-07-28 线缆协议上（这两个动词已不存在）该请求会得到-32601，即Method not found。2026 时代的替代方案是subscriptions/listen流，MCPServer确实提供它——此时server_capabilities.resources.subscribe为True——用client.listen(...)消费它的方法见本节的 订阅。

Prompts：列出与渲染

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)

代码位于 docs_src/client/tutorial005.py。list_prompts()告诉你服务器提供什么、每个 prompt 需要什么参数：

prompt.name # 'recommend' prompt.title # 'Recommend a book' prompt.arguments # [PromptArgument(name='genre', required=True)]

get_prompt(name, arguments)负责渲染。参数字典是str -> str：prompt 参数永远是字符串。结果在messages中，是PromptMessage列表，每个含role与一个content块：

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

宿主把这些消息原样交给模型即可——这就是该功能的全部。

Completions：自动补全

带 completion 处理器的服务器可以在用户输入过程中自动补全 prompt 参数和资源模板参数：

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)

代码位于 docs_src/client/tutorial006.py。两个关键点：

• ref指明你在补全哪个prompt 或模板：PromptReference或ResourceTemplateReference；
• argument为{"name": ..., "value": ...}：参数名与用户截至目前输入的内容。

答案在result.completion.values中。输入"p"，服务器返回['poetry']。服务器端实现、以及处理器如何利用其他已填参数收窄建议，见 Completions。

分页：cursor=与next_cursor

每个list_*方法都接受cursor=关键字参数，每个结果都携带next_cursor。当next_cursor为None时，说明已取完全部数据：

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)

代码位于 docs_src/client/tutorial007.py。list_all_tools这个循环对任何服务器都是正确的：MCPServer一次性返回全部数据，因此next_cursor为None、循环只执行一次——这就是大多数代码从不写分页的原因。真正做分页的服务器、以及游标遵循的规则，见 分页。

在测试中使用

本页每个client.py都是通过 HTTP 连到server.py的。而在测试中，你跳过网络，直接把服务器对象交给Client：

from server import mcp client = Client(mcp)

无进程、无端口，且上面提到的每个方法行为完全一致。专为此设计的构造器参数是Client(mcp, raise_exceptions=True)——它只对进程内连接生效。测试 一节解释了它并围绕它构建了完整模式。

小结

• Client(x)：对 URL 字符串走 Streamable HTTP，对StdioServerParameters启动子进程，对 transport 直接进入，在测试中则接收服务器对象本身；
• 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