企业官网建设流程全解析

python-sdk 低层 Server 实战：用 MCP 裸协议对象手工构建服务器

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

@mcp.tool()只是 python-sdk（Model 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 未定义方法）的完整降级方案。

为什么需要低层Server

MCPServer构建在Server之上，二者不是竞争关系，而是分层关系。正如MCPServer是"装饰器 + 类型注解"层、Server是它底下的 Starlette 一样，MCPServer内部会构造一个Server并向其注册与本文完全相同的处理器（handler）。当便捷层碍事时，你才需要降级到低层：

• 你需要发出精确的 schema（从文件加载、由数据库生成），而不是从 Python 函数签名推导出的 schema；
• 你需要对结果拥有完全控制：_meta、is_error、structured_content的每一个键；
• 你需要处理 MCP 规范没有定义的方法。

其余场景，继续使用MCPServer即可。

手写同一个工具：去糖后的完整 API

search_books这个工具在 docs/servers/tools.md 中只用九行@mcp.tool()就能实现；下面是不含任何语法糖的版本，完整源码见 docs_src/lowlevel/tutorial001.py：

from mcp.server import Server, ServerRequestContext from mcp.types import ( CallToolRequestParams, CallToolResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool, ) SEARCH_BOOKS = Tool( name="search_books", description="Search 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 = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})." return CallToolResult(content=[TextContent(type="text", text=text)]) server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool) app = server.streamable_http_app()

与高层版本相比，有三处变化，而这三点合起来就是低层 API 的全部：

• 处理器是构造器参数：on_list_tools=与on_call_tool=直接传入Server(...)。这一层没有装饰器，且每个处理器都是同一形态：async (ctx, params) -> result。
• 你亲自编写输入 schema：Tool.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/mcp：

import 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(type='text', text="Found 3 books matching 'dune' (showing up to 5).", annotations=None, meta=None)]

与@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返回的就是你写的那份字面 dict，test_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_exceptions=True会改抛真实异常，参见 docs/get-started/testing.md。）

这一点可以推广：从低层处理器抛出的异常永远是协议错误，永远不会变成is_error=True的工具结果。如果你希望模型能读到失败并自我纠正，就必须自己校验params.arguments，然后返回CallToolResult(content=[TextContent(...)], is_error=True)。两类失败方式的完整讨论见 docs/servers/handling-errors.md。

对应的测试test_arguments_are_not_validated_against_your_schema（tests/docs_src/test_lowlevel.py）验证了缺参调用确实穿透到处理器内部并在那里炸掉，客户端收到ErrorData(code=INTERNAL_ERROR, message="Internal server error")。

两个工具，一个处理器

on_call_tool是整个服务器所有工具的唯一入口，你需要根据params.name自行路由（源码见 docs_src/lowlevel/tutorial002.py）：

async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult: args = params.arguments or {} if params.name == "search_books": text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})." elif params.name == "add_book": text = f"Added {args['title']!r} by {args['author']} ({args['year']})." else: raise ValueError(f"Unknown tool: {params.name}") return CallToolResult(content=[TextContent(type="text", text=text)])

• 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.py）：

SEARCH_BOOKS = Tool( name="search_books", description="Search 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(type="text", text=f"Found 3 books matching {args['query']!r}.")], structured_content=data, ) server = Server("Bookshop", version="2.0.0", on_list_tools=list_tools, on_call_tool=call_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_schema，call_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-12

input_schema与output_schema都是 JSON Schema。MCP 规范固定了方言：没有$schema键的 schema 一律视为JSON Schema 2020-12。MCPServer生成的 schema 正是依赖这一默认值（Pydantic 写出 2020-12 并省略该键），手工编写的 dict 同样受此约束，因此完整的 2020-12 词汇表都可用（示例见 docs_src/lowlevel/tutorial007.py）：

FIND_BOOK = Tool( name="find_book", description="Find 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.py）：

return CallToolResult( content=[TextContent(type="text", text=f"Found 3 books matching {args['query']!r}.")], structured_content=data, _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_prompts，prompts就会出现；传入on_completion，completions就会出现。

对比之下，MCPServer无论你是否注册了工具、资源、提示词，都会永远公告这三者，因为它的管理器始终存在。在低层，"声明"就是"构造器调用"。测试test_only_the_handlers_you_passed_become_capabilities精确断言了client.server_capabilities只包含{"tools": {"list_changed": False}}。

生命周期泛型Server[T]

Server以其生命周期（lifespan）产出的类型为泛型参数。只需注解一次，这个对象就在它出现的任何地方都带有类型（示例见 docs_src/lowlevel/tutorial005.py）：

@dataclass 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 = f"Found {len(matches)} books: {', '.join(matches)}." return CallToolResult(content=[TextContent(type="text", text=text)]) server = Server("Bookshop", lifespan=lifespan, on_list_tools=list_tools, on_call_tool=call_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.py）：

class ReindexParams(RequestParams): full: bool = False class ReindexResult(BaseModel): indexed: int async def reindex(ctx: ServerRequestContext, params: ReindexParams) -> ReindexResult: return ReindexResult(indexed=3) server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_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或None，SDK 会把它序列化进 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_audience=server.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_error=True的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