尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

python-sdk 客户端开发实战:用 Python `Client` 全面掌握 MCP 协议的每个动词

python-sdk 客户端开发实战:用 Python `Client` 全面掌握 MCP 协议的每个动词 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-sdkModel 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, 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)])这段完整代码位于 docs_src/client/tutorial001.py。它同时注册了工具、资源、prompt 与 completion 处理器是本文所有客户端示例的对端。启动命令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)这段代码来自 docs_src/client/tutorial001_client.py。三个要点Client(http://localhost:8000/mcp)接收的是一个URL因此通过Streamable HTTP传输连接到刚启动的服务器async with就是生命周期进入时连接并协商协议退出时断开进入块之后连接事实已经以普通属性plain properties的形式就绪直接读取即可。可以传给Client的四种参数Client只接收一个位置参数并根据其类型自动解析出对应的传输方式传入类型传输方式典型场景URL 字符串如Client(http://localhost:8000/mcp)Streamable HTTP部署在 HTTP 服务后面的远程服务器StdioServerParameters本地子进程stdio通过 stdin/stdout 与本地进程对话任意 transport直接进入如streamable_http_client(url, http_client...)围绕自有 HTTP 客户端定制传输MCPServer或底层Server实例进程内连接in-process测试无子进程、无端口除传输方式外本页其余内容对四种情况完全一致。关于自定义 header、子进程参数、超时与Transport协议本身的细节见 客户端传输。其中进程内模式是测试的基石测试 一节专门围绕它构建。已连接客户端上有什么进入async with块后四个只读属性即被填充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会**探测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)执行工具并返回CallToolResultimport 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(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一个返回值三处可读各有各的消费方。从 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_errorTrue的结果返回。验证一下让lookup_book查找Solaris目录中不存在的书名函数会抛出ToolError但调用仍正常返回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_errorTrue且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(namegenre, requiredTrue)]get_prompt(name, arguments)负责渲染。参数字典是str - strprompt 参数永远是字符串。结果在messages中是PromptMessage列表每个含role与一个content块message.role # user message.content # TextContent(typetext, textRecommend 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( refPromptReference(typeref/prompt, namerecommend), argument{name: genre, value: p}, ) print(result.completion.values) if __name__ __main__: anyio.run(main)代码位于 docs_src/client/tutorial006.py。两个关键点ref指明你在补全哪个prompt 或模板PromptReference或ResourceTemplateReferenceargument为{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(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)代码位于 docs_src/client/tutorial007.py。list_all_tools这个循环对任何服务器都是正确的MCPServer一次性返回全部数据因此next_cursor为None、循环只执行一次——这就是大多数代码从不写分页的原因。真正做分页的服务器、以及游标遵循的规则见 分页。在测试中使用本页每个client.py都是通过 HTTP 连到server.py的。而在测试中你跳过网络直接把服务器对象交给Clientfrom server import mcp client Client(mcp)无进程、无端口且上面提到的每个方法行为完全一致。专为此设计的构造器参数是Client(mcp, raise_exceptionsTrue)——它只对进程内连接生效。测试 一节解释了它并围绕它构建了完整模式。小结Client(x)对 URL 字符串走 Streamable HTTP对StdioServerParameters启动子进程对 transport 直接进入在测试中则接收服务器对象本身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),仅供参考
返回列表