python-sdk 低层 Server 实战用 MCP 裸协议对象手工构建服务器【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdkmcp.tool()只是 python-sdkModel Context Protocol 官方 Python SDK提供的一层语法糖。在它之下还有一层名为Server的服务器类直接以 MCP 协议对象为输入你把Tool、CallToolResult等协议对象交给它它会原封不动地放到链路上。本文以 docs/advanced/low-level-server.md 为主线结合 docs_src/lowlevel/ 下的完整可运行示例与 src/mcp/server/lowlevel/server.py 源码系统讲解如何不借助装饰器、不依赖类型注解手工编写input_schema、手工构造返回结果、注册自定义方法并理解校验、_meta、能力声明、生命周期泛型等在低层的真实行为。读完你将掌握在便捷层无法满足精确控制需求时如精确 schema、_meta完全控制、MCP 未定义方法的完整降级方案。为什么需要低层ServerMCPServer构建在Server之上二者不是竞争关系而是分层关系。正如MCPServer是装饰器 类型注解层、Server是它底下的 Starlette 一样MCPServer内部会构造一个Server并向其注册与本文完全相同的处理器handler。当便捷层碍事时你才需要降级到低层你需要发出精确的 schema从文件加载、由数据库生成而不是从 Python 函数签名推导出的 schema你需要对结果拥有完全控制_meta、is_error、structured_content的每一个键你需要处理 MCP 规范没有定义的方法。其余场景继续使用MCPServer即可。手写同一个工具去糖后的完整 APIsearch_books这个工具在 docs/servers/tools.md 中只用九行mcp.tool()就能实现下面是不含任何语法糖的版本完整源码见 docs_src/lowlevel/tutorial001.pyfrom mcp.server import Server, ServerRequestContext from mcp.types import ( CallToolRequestParams, CallToolResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool, ) SEARCH_BOOKS Tool( namesearch_books, descriptionSearch the catalog by title or author., input_schema{ type: object, properties: {query: {type: string}, limit: {type: integer}}, required: [query, limit], }, ) async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) - ListToolsResult: return ListToolsResult(tools[SEARCH_BOOKS]) async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) - CallToolResult: args params.arguments or {} text fFound 3 books matching {args[query]!r} (showing up to {args[limit]}). return CallToolResult(content[TextContent(typetext, texttext)]) server Server(Bookshop, on_list_toolslist_tools, on_call_toolcall_tool) app server.streamable_http_app()与高层版本相比有三处变化而这三点合起来就是低层 API 的全部处理器是构造器参数on_list_tools与on_call_tool直接传入Server(...)。这一层没有装饰器且每个处理器都是同一形态async (ctx, params) - result。你亲自编写输入 schemaTool.input_schema就是一个普通的 JSON Schemadict。没有人从类型注解推导它因为这里根本没有可供推导的类型注解。你亲手构造结果CallToolResult(content[TextContent(...)])逐字手写。没有包装、没有转换、没有从返回注解推断任何东西。params是解析后的请求CallToolRequestParams提供.name与.arguments。ctx是ServerRequestContext可访问ctx.session向客户端回话、ctx.lifespan_context、ctx.request_id以及ctx.meta请求入站的_meta。从源码看Server的完整构造器签名位于 src/mcp/server/lowlevel/server.py可以看到它接受name、可选version、lifespan以及一整套on_*处理器参数on_list_tools、on_call_tool、on_list_resources、on_list_prompts、on_completion、on_ping等且每个处理器都声明为可空默认None——这意味着你注册了什么服务器就拥有什么。亲自运行它mcp dev与mcp run只接受MCPServer所以低层服务器需要你自己托管。server.py最后一行server.streamable_http_app()从低层服务器构建出一个普通的 Starlette ASGI 应用从源码看streamable_http_app 返回的正是与MCPServer相同的应用用 uvicorn 即可启动uvicorn server:app --port 8000然后用 Inspector 或任意客户端指向http://localhost:8000/mcpimport asyncio from mcp import Client async def main() - None: async with Client(http://localhost:8000/mcp) as client: result await client.call_tool(search_books, {query: dune, limit: 5}) print(result.content) asyncio.run(main())输出[TextContent(typetext, textFound 3 books matching dune (showing up to 5)., annotationsNone, metaNone)]与mcp.tool()版本产生的文本完全一致。诚实地讲有两个差异result.structured_content为None。高层服务器会把- str自动包装成{result: ...}在这里你没有构造的东西就没有人替你构造。list_tools返回的是你亲手输入的 schema逐字符一致。高层版本在每个属性上有title: Query、根上有title: search_booksArguments——这些是 Pydantic 的产物。在低层只要出现在链路上就必然是你自己放上去的。在测试中你可以完全跳过 uvicorn 与端口Client(server)可以在同进程内接受低层Server就像接受MCPServer一样。对应的测试见 tests/docs_src/test_lowlevel.py例如test_the_input_schema_on_the_wire_is_the_dict_you_wrote断言tools/list返回的就是你写的那份字面 dicttest_the_last_line_is_an_asgi_app_uvicorn_can_serve则验证app是一个只有/mcp单一路由的 Starlette 应用。更多细节见 docs/get-started/testing.md。什么都不会替你校验MCPServer会在你的函数运行之前就拒绝非法参数——它会把调用与自身生成的 schema 进行校验见 docs/servers/tools.md。Server不会这么做。你的input_schema只是向客户端公告advertised却从不应用applied到params.arguments上。试着不带limit调用search_books你的args[limit]会抛出KeyError客户端看到的是MCPError: Internal server error这是一条 JSON-RPC 错误错误码-32603消息被刻意写得通用SDK 不会把 traceback 泄露给远端调用者。模型永远不知道自己错在哪里因此也无法重试。在测试中raise_exceptionsTrue会改抛真实异常参见 docs/get-started/testing.md。这一点可以推广从低层处理器抛出的异常永远是协议错误永远不会变成is_errorTrue的工具结果。如果你希望模型能读到失败并自我纠正就必须自己校验params.arguments然后返回CallToolResult(content[TextContent(...)], is_errorTrue)。两类失败方式的完整讨论见 docs/servers/handling-errors.md。对应的测试test_arguments_are_not_validated_against_your_schematests/docs_src/test_lowlevel.py验证了缺参调用确实穿透到处理器内部并在那里炸掉客户端收到ErrorData(codeINTERNAL_ERROR, messageInternal server error)。两个工具一个处理器on_call_tool是整个服务器所有工具的唯一入口你需要根据params.name自行路由源码见 docs_src/lowlevel/tutorial002.pyasync def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) - CallToolResult: args params.arguments or {} if params.name search_books: text fFound 3 books matching {args[query]!r} (showing up to {args[limit]}). elif params.name add_book: text fAdded {args[title]!r} by {args[author]} ({args[year]}). else: raise ValueError(fUnknown tool: {params.name}) return CallToolResult(content[TextContent(typetext, texttext)])list_tools负责公告这两个工具call_tool根据名称分发。else分支很重要Server会毫不犹豫地把一个你从未列出过的名称对应的tools/call直接送进你的处理器。在那里抛异常会把这次调用变成与上文相同的-32603错误测试test_an_unknown_tool_name_becomes_a_protocol_error_not_a_tool_error验证了这一点。手工结构化输出在Tool上声明output_schema并把structured_content放到结果里两者都由你掌控源码见 docs_src/lowlevel/tutorial003.pySEARCH_BOOKS Tool( namesearch_books, descriptionSearch the catalog by title or author., input_schema{ type: object, properties: {query: {type: string}, limit: {type: integer}}, required: [query, limit], }, output_schema{ type: object, properties: {matches: {type: integer}, query: {type: string}}, required: [matches, query], }, ) async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) - CallToolResult: args params.arguments or {} data {matches: 3, query: args[query]} return CallToolResult( content[TextContent(typetext, textfFound 3 books matching {args[query]!r}.)], structured_contentdata, ) server Server(Bookshop, version2.0.0, on_list_toolslist_tools, on_call_toolcall_tool)调用它结果会同时携带两种表示{ content: [{type: text, text: Found 3 books matching dune.}], structuredContent: {matches: 3, query: dune}, isError: false, resultType: complete, _meta: {io.modelcontextprotocol/serverInfo: {name: Bookshop, version: 2.0.0}} }其中的_meta块是服务器的身份戳SDK 会把它加到每一个 2026 时代的2026-era结果上version取自构造器没有设置 version 的服务器会输出空字符串。不希望暴露身份的服务器可以用一个中间件移除该键——中间件对它返回的结果拥有完全控制权。服务器永远不会比对这两个字段。但本 SDK 的Client会如果你返回的structured_content不满足自己声明的output_schemacall_tool会抛出以Invalid structured content returned by tool search_books开头、随后引用jsonschema失败详情的RuntimeError。承诺一个 schema 不花任何成本兑现它则是你的责任。测试test_the_client_checks_the_schema_you_promised构造了一个structured_content{matches: three}的违约服务器来验证这一点。返回类型与 schema 的完整阶梯见 docs/servers/structured-output.md。方言是 JSON Schema 2020-12input_schema与output_schema都是 JSON Schema。MCP 规范固定了方言没有$schema键的 schema 一律视为JSON Schema 2020-12。MCPServer生成的 schema 正是依赖这一默认值Pydantic 写出 2020-12 并省略该键手工编写的 dict 同样受此约束因此完整的 2020-12 词汇表都可用示例见 docs_src/lowlevel/tutorial007.pyFIND_BOOK Tool( namefind_book, descriptionFind one book by ISBN, or by title and author., input_schema{ type: object, properties: { isbn: {type: string, pattern: ^[0-9]{13}$}, title: {type: string}, author: {type: string}, }, oneOf: [{required: [isbn]}, {required: [title, author]}], additionalProperties: False, }, )input_schema的根必须是type: object。除此之外oneOf、additionalProperties、anyOf、if/then/else、prefixItems、带本地$ref的$defs以及其余 2020-12 关键字都会原样送达客户端。不需要任何$schema键。只有当你想选择更旧的 draft 版本时才需要添加本 SDK 的Client在把structured_content与工具的output_schema比对时会根据$schema挑选校验器缺省时使用 2020-12。_meta给应用而不是给模型content是模型读取的那部分答案structured_content是同一答案的类型化数据形态_meta是第三条通道——跟随结果一起到达客户端应用、却不属于答案本身的数据。用它来携带记录 ID、trace ID以及一切你的 UI 需要、而 prompt 不需要的东西示例见 docs_src/lowlevel/tutorial004.pyreturn CallToolResult( content[TextContent(typetext, textfFound 3 books matching {args[query]!r}.)], structured_contentdata, _meta{bookshop/record_ids: [bk_17, bk_42, bk_99]}, )你在服务器端以_meta构造它这是它在链路上的名字客户端以result.meta读回。请为你的键加命名空间前缀如bookshop/record_ids。io.modelcontextprotocol/*命名空间下的键由协议保留上文的身份戳io.modelcontextprotocol/serverInfo即属此类。_meta是你与客户端应用之间的约定而不是关于什么内容会到达模型的保证。宿主决定它渲染什么。永远不要在工具结果的任何部分放入秘密。测试test_meta_reaches_the_client_application验证了_meta会以result.meta读回、以_meta序列化并且与服务器身份戳共享同一块_meta而不互相覆盖。能力声明跟随你的处理器一个Server只会公告你为它提供了处理器的那些方法族。上面的Bookshop只传了on_list_tools与on_call_tool因此连接它的客户端看到的能力是{tools: {listChanged: false}}没有resources、没有prompts——因为背后没有任何东西支撑它们。传入on_list_promptsprompts就会出现传入on_completioncompletions就会出现。对比之下MCPServer无论你是否注册了工具、资源、提示词都会永远公告这三者因为它的管理器始终存在。在低层声明就是构造器调用。测试test_only_the_handlers_you_passed_become_capabilities精确断言了client.server_capabilities只包含{tools: {list_changed: False}}。生命周期泛型Server[T]Server以其生命周期lifespan产出的类型为泛型参数。只需注解一次这个对象就在它出现的任何地方都带有类型示例见 docs_src/lowlevel/tutorial005.pydataclass class Catalog: books: list[str] def search(self, query: str) - list[str]: return [title for title in self.books if query.lower() in title.lower()] asynccontextmanager async def lifespan(server: Server[Catalog]) - AsyncIterator[Catalog]: yield Catalog(books[Dune, Dune Messiah, Children of Dune]) async def call_tool(ctx: ServerRequestContext[Catalog], params: CallToolRequestParams) - CallToolResult: matches ctx.lifespan_context.search((params.arguments or {})[query]) text fFound {len(matches)} books: {, .join(matches)}. return CallToolResult(content[TextContent(typetext, texttext)]) server Server(Bookshop, lifespanlifespan, on_list_toolslist_tools, on_call_toolcall_tool)生命周期是一个Callable[[Server[Catalog]], AbstractAsyncContextManager[Catalog]]把asynccontextmanager用在async生成器上得到的就是它这与源码中 Server.init的lifespan参数签名完全对应。它yield出来的东西会成为ctx.lifespan_context由于处理器被注解为ServerRequestContext[Catalog].search(...)可以自动补全并通过类型检查。服务器启动时进入一次停止时退出一次。启动、关闭以及MCPServer版本的同一概念见 docs/handlers/lifespan.md。不带lifespan时ctx.lifespan_context是一个空dict。自定义方法add_request_handler构造器覆盖了 MCP 定义的方法add_request_handler则覆盖其余一切示例见 docs_src/lowlevel/tutorial006.pyclass ReindexParams(RequestParams): full: bool False class ReindexResult(BaseModel): indexed: int async def reindex(ctx: ServerRequestContext, params: ReindexParams) - ReindexResult: return ReindexResult(indexed3) server Server(Bookshop, on_list_toolslist_tools, on_call_toolcall_tool) server.add_request_handler(bookshop/reindex, ReindexParams, reindex)第一个参数是方法名字符串。通知有它的孪生兄弟add_notification_handler见 src/mcp/server/lowlevel/server.py。通知处理器的触发范围是 stdio 以及握手时代handshake-era的 HTTP 连接在2026-07-28版本的 Streamable HTTP 路径上客户端发来的通知 POST 只会收到202确认而不会被分发——因为该修订版在 HTTP 上没有定义任何客户端到服务器的通知。params_type是入站params在你的处理器运行之前被校验所依据的模型自定义方法确实享有工具所没有的校验能力。继承RequestParams_meta字段就会像其他任何方法一样被解析。处理器可以返回BaseModel、dict或NoneSDK 会把它序列化进 JSON-RPC 结果。测试test_add_request_handler_registers_a_method_the_constructor_does_not_know验证了处理器与其params_type确实被登记进注册表。一个诚实的告诫高层Client只为 MCP 定义的方法提供动词所以不会有client.reindex()。自定义方法面向的是已知其存在的对端你自己同时交付的客户端或者另一个讲 JSON-RPC 的你自己的服务。有一种方法你不能据为己有ValueError: initialize is handled by the server runner and cannot be overridden; use Server.middleware to observe or wrap initialization握手handshake属于 runner。server/discover、ping以及其他所有内置方法则任你替换。测试test_initialize_is_reserved精确复现了这条ValueError。错误消息中提到的Server.middleware会包裹每一条入站消息包括initialize。如果你想要的是观察或改写流量、而非回应一个新方法请从 docs/advanced/middleware.md 开始。其他处理器一览其余每个处理器都对应一个你现在已经掌握词汇的概念各有专门页面on_call_tool、on_get_prompt、on_read_resource可以返回InputRequiredResult来代替正常结果从而暂停调用并向客户端索取输入见 docs/handlers/multi-round-trip.md。这一层忠实于不替你安装任何东西的原则MCPServer默认封缄requestState而在这里你设置的request_state会原样穿越链路直到你主动用一行代码server.middleware.append(RequestStateBoundary(RequestStateSecurity(keys[...]), default_audienceserver.name))显式加入两个名字都从mcp.server.request_state导入即可获得与MCPServer完全相同的封缄与校验见 docs/handlers/multi-round-trip.md 中ProtectingrequestState一节。on_list_resources、on_read_resource、on_list_prompts、on_get_prompt、on_completion对其他原语保持同一形态(ctx, params) - result。on_subscriptions_listen服务于2026-07-28版本的subscriptions/listen流。传入一个构建在SubscriptionBus之上的ListenHandler并从其他处理器向该总线发布事件完整组合见 docs/handlers/subscriptions.md。server.streamable_http_app()返回与MCPServer相同的 Starlette 应用按 docs/run/index.md 部署任意 ASGI 应用的方式部署它即可。这一层没有server.run(transport...)server.run(read_stream, write_stream, server.create_initialization_options())见 src/mcp/server/lowlevel/server.py在一对流上驱动一条连接这一行就是全部故事。小结低层Server以on_*构造器参数接收处理器每个处理器都是async (ctx, params) - result。你编写input_schemadict手工构造CallToolResult。没有任何东西被推导、包装或替你校验。处理器中的异常是-32603协议错误模型可读的工具错误是你主动返回的、带is_errorTrue的CallToolResult。结果上的_meta面向客户端应用而不是模型。Server[T]以其生命周期产出的类型为泛型参数ctx.lifespan_context是一个带类型的T。add_request_handler(method, params_type, handler)服务任意方法initialize保留给 runner。Server公告的能力由你注册的处理器推导而来。客户端之所以以完全相同的方式对待两种服务器是因为它们本来就是同一个协议——这正是全部要点所在。再往下一层就不再是类了那是 docs/advanced/middleware.md。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考