ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

mcp-for-beginners 进阶指南:使用 MCP 低层服务器(Low-Level Server)构建可扩展架构

mcp-for-beginners 进阶指南:使用 MCP 低层服务器(Low-Level Server)构建可扩展架构 教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载本篇文章基于 mcp-for-beginners 开源课程 10-advanced 章节 展开核心讲解 MCP SDK 中普通服务器Regular Server与低层服务器Low-Level Server的区别以及如何用低层服务器的每个特性类型仅两个处理器模式配合 PydanticPython与 ZodTypeScript校验构建一个按 tools / resources / prompts 目录组织、易于持续扩展的服务器架构。读完本篇你将能够独立编写一个低层 MCP 服务器完成工具列表与工具调用的协议处理、入参校验并理解为什么这种模式更利于架构治理。为什么需要低层服务器MCP SDK 中暴露了两类服务器普通服务器Regular Server例如 Python 的FastMCP、TypeScript 的McpServer与低层服务器Low-Level Server。日常开发中我们通常使用普通服务器只需向它注册功能即可。但在一些场景下低层服务器反而更合适更好的架构。普通服务器和低层服务器都能构建整洁架构但可以认为低层服务器下组织代码更直观——因为功能注册被集中收拢为少量处理器。特性可用性。部分高级特性只能通过低层服务器使用。后续章节涉及的 Elicitation 与 legacy Sampling 特性即是如此其中 Sampling 在 MCP 规范2026-07-28版本中已被标记为废弃。普通服务器的注册方式以添加一个两数相加工具为例。PythonFastMCPmcp FastMCP(Demo) # Add an addition tool mcp.tool() def add(a: int, b: int) - int: Add two numbers return a bTypeScriptMcpServerconst server new McpServer({ name: demo-server, version: 1.0.0 }); // Add an addition tool server.registerTool(add, { title: Addition Tool, description: Add two numbers, inputSchema: { a: z.number(), b: z.number() } }, async ({ a, b }) ({ content: [{ type: text, text: String(a b) }] }) );可以看到普通服务器要求你显式地逐个注册每个工具、资源或提示词。这本身没有任何问题但当特性数量增长后所有注册逻辑会逐渐堆积在入口文件中。低层服务器的思维转变低层服务器要求换一种思考方式不再逐个注册工具而是为每种特性类型tools、resources、prompts各实现两个处理器列出所有工具一个函数负责所有列出工具的请求处理工具调用只有一个函数负责响应所有调用工具的请求。这听起来工作量更少不是吗你不需要注册工具只需要保证当客户端请求列出工具时工具出现在列表中当有调用请求进来时工具被正确分派执行。低层服务器的两个核心处理器1. 列出工具List ToolsPython使用server.list_tools()装饰器server.list_tools() async def handle_list_tools() - list[types.Tool]: List available tools. return [ types.Tool( nameadd, descriptionAdd two numbers, inputSchema{ type: object, properties: { a: {type: number, description: number to add}, b: {type: number, description: number to add} }, required: [query], }, ) ]TypeScript调用server.setRequestHandler并传入ListToolsRequestSchemaserver.setRequestHandler(ListToolsRequestSchema, async (request) { // Return the list of registered tools return { tools: [{ name: add, description: Add two numbers, inputSchema: { type: object, properties: { a: {type: number, description: number to add}, b: {type: number, description: number to add} }, required: [query], } }] }; });这里返回的是一个特性列表每个条目包含name、description、inputSchema等字段以符合返回类型的要求。这意味着工具的定义数据可以与服务器主文件分离——我们可以把所有工具放进tools目录、资源放进resources目录、提示词放进prompts目录项目突然就能组织成这样app --| tools ----| add ----| substract --| resources ----| products ----| schemas --| prompts ----| product-description2. 调用工具Call Tool调用工具也是同一个思路只用一个处理器根据请求中的工具名分派到对应实现。Python使用server.call_tool()装饰器server.call_tool() async def handle_call_tool( name: str, arguments: dict[str, str] | None ) - list[types.TextContent]: # tools is a dictionary with tool names as keys if name not in tools.tools: raise ValueError(fUnknown tool: {name}) tool tools.tools[name] result default try: result await toolhandler except Exception as e: raise ValueError(fError calling tool {name}: {str(e)}) return [ types.TextContent(typetext, textstr(result)) ]TypeScript同样通过CallToolRequestSchema注册处理器server.setRequestHandler(CallToolRequestSchema, async (request) { const { params: { name } } request; let tool tools.find(t t.name name); if(!tool) { return { error: { code: tool_not_found, message: Tool ${name} not found. } }; } // args: request.params.arguments // TODO call the tool, return { content: [{ type: text, text: Tool ${name} called with arguments: ${JSON.stringify(input)}, result: ${JSON.stringify(result)} }] }; });从以上代码可以看出我们需要解析出要调用的工具名、携带的参数然后继续执行工具调用。用校验进一步完善方案到目前为止你已经看到注册工具、资源、提示词的全部工作可以替换为每种特性类型各两个处理器。接下来还需要做一件事——入参校验确保工具被以正确的参数调用。每种运行时都有自己的解决方案Python 使用 PydanticTypeScript 使用 Zod。整体思路是把创建特性工具、资源或提示词的逻辑移动到其专属目录增加一种机制校验调用工具这类入站请求的参数合法性。创建特性Feature文件每个特性文件需要包含该特性类型的必填字段tools、resources、prompts 的字段略有差异。Python用 Pydantic 定义输入模型再把工具定义与处理函数打包成字典。# schema.py from pydantic import BaseModel class AddInputModel(BaseModel): a: float b: float # add.py from .schema import AddInputModel async def add_handler(args) - float: try: # Validate input using Pydantic model input_model AddInputModel(**args) except Exception as e: raise ValueError(fInvalid input: {str(e)}) # TODO: add Pydantic, so we can create an AddInputModel and validate args Handler function for the add tool. return float(input_model.a) float(input_model.b) tool_add { name: add, description: Adds two numbers, input_schema: AddInputModel, handler: add_handler }这里做了两件事在schema.py中用 Pydantic 创建包含字段a、b的AddInputModel在add.py中尝试把入站请求解析为AddInputModel若参数不匹配就会抛出异常# add.py try: # Validate input using Pydantic model input_model AddInputModel(**args) except Exception as e: raise ValueError(fInvalid input: {str(e)})你可以自主选择把这段解析逻辑放在工具调用处理器中还是放在各个 handler 函数内部。TypeScript在server.ts的调用处理器里用 Zod 解析参数并处理两种失败情形工具不存在、参数非法。// server.ts server.setRequestHandler(CallToolRequestSchema, async (request) { const { params: { name } } request; let tool tools.find(t t.name name); if (!tool) { return { error: { code: tool_not_found, message: Tool ${name} not found. } }; } const Schema tool.rawSchema; try { const input Schema.parse(request.params.arguments); // ts-ignore const result await tool.callback(input); return { content: [{ type: text, text: Tool ${name} called with arguments: ${JSON.stringify(input)}, result: ${JSON.stringify(result)} }] }; } catch (error) { return { error: { code: invalid_arguments, message: Invalid arguments for tool ${name}: ${error instanceof Error ? error.message : String(error)} } }; } }); // schema.ts import { z } from zod; export const MathInputSchema z.object({ a: z.number(), b: z.number() }); // add.ts import { Tool } from ./tool.js; import { MathInputSchema } from ./schema.js; import { zodToJsonSchema } from zod-to-json-schema; export default { name: add, rawSchema: MathInputSchema, inputSchema: zodToJsonSchema(MathInputSchema), callback: async ({ a, b }) { return { content: [{ type: text, text: String(a b) }] }; } } as Tool;在统一处理所有工具调用的 handler 中我们先尝试把入站请求解析为该工具定义的 schemaconst Schema tool.rawSchema; try { const input Schema.parse(request.params.arguments);如果解析成功再调用真正的工具实现const result await tool.callback(input);可以看出这种方案塑造了很好的架构一切都有其归属位置server.ts变成一个非常小的文件只负责装配请求处理器每个特性都位于各自目录tools/、resources/、prompts/中。动手练习创建一个低层服务器本练习的目标是创建一个能够处理工具列表与工具调用的低层服务器实现一个可以持续扩展的架构加入校验确保工具调用得到正确的参数验证。-1- 创建架构首先需要一个有助于扩展的目录结构。Pythonserver.py --| tools ----| __init__.py ----| add.py ----| schema.py client.pyTypeScriptserver.ts --| tools ----| add.ts ----| schema.ts client.ts这套架构保证了我们可以在tools目录中轻松添加新工具。你也可以遵循同样的思路为 resources 和 prompts 添加子目录。-2- 创建工具工具文件放在tool子目录中。Pythontools/add.py定义 name、description用 Pydantic 定义输入 schema以及一个在工具被调用时执行的 handler最后通过tool_add字典把所有这些属性暴露出来。from .schema import AddInputModel async def add_handler(args) - float: try: # Validate input using Pydantic model input_model AddInputModel(**args) except Exception as e: raise ValueError(fInvalid input: {str(e)}) # TODO: add Pydantic, so we can create an AddInputModel and validate args Handler function for the add tool. return float(input_model.a) float(input_model.b) tool_add { name: add, description: Adds two numbers, input_schema: AddInputModel, handler: add_handler }配套的tools/schema.py定义工具使用的输入 schemafrom pydantic import BaseModel class AddInputModel(BaseModel): a: float b: float还需要填充tools/__init__.py让tools目录被当作一个模块并对外暴露其内部的模块from .add import tool_add tools { tool_add[name] : tool_add }随着工具增多可以持续往这个文件里追加条目。TypeScriptsrc/tools/add.ts创建包含以下属性的字典name工具名称rawSchemaZod schema用于校验对该工具的入站调用请求inputSchema供处理器使用的 JSON Schema由zodToJsonSchema转换而来callback实际调用工具的函数。import { Tool } from ./tool.js; import { MathInputSchema } from ./schema.js; import { zodToJsonSchema } from zod-to-json-schema; export default { name: add, rawSchema: MathInputSchema, inputSchema: zodToJsonSchema(MathInputSchema), callback: async ({ a, b }) { return { content: [{ type: text, text: String(a b) }] }; } } as Tool;Tool接口用于把这个字典转换为 MCP 服务器 handler 可接受的类型import { z } from zod; export interface Tool { name: string; inputSchema: any; rawSchema: z.ZodTypeAny; callback: (args: z.inferz.ZodTypeAny) Promise{ content: { type: string; text: string }[] }; }schema.ts集中存放每个工具的输入 schema目前只有一个随着工具增加可以不断添加条目import { z } from zod; export const MathInputSchema z.object({ a: z.number(), b: z.number() });-3- 处理工具列表在服务器文件中注册列出工具的请求处理器。Python添加server.list_tools装饰器与handle_list_tools实现函数。注意每个工具都需要 name、description 和 inputSchema# code omitted for brevity from tools import tools server.list_tools() async def handle_list_tools() - list[types.Tool]: tool_list [] print(tools) for tool in tools.values(): tool_list.append( types.Tool( nametool[name], descriptiontool[description], inputSchemapydantic_to_json(tool[input_schema]), ) ) return tool_listTypeScript调用server.setRequestHandler传入与目标匹配的 schema这里是ListToolsRequestSchema// index.ts import addTool from ./add.js; import subtractTool from ./subtract.js; import {server} from ../server.js; import { Tool } from ./tool.js; export let tools: ArrayTool []; tools.push(addTool); tools.push(subtractTool); // server.ts // code omitted for brevity import { tools } from ./tools/index.js; server.setRequestHandler(ListToolsRequestSchema, async (request) { // Return the list of registered tools return { tools: tools }; });这样列出工具这一环就完成了。-4- 处理工具调用调用工具需要再注册一个请求处理器专门处理指定调用哪个特性、携带什么参数的请求。Python使用server.call_tool装饰器实现handle_call_tool。函数内需要解析出工具名与参数并确保参数对目标工具有效——可以在这个函数中校验也可以下放到具体工具内部校验server.call_tool() async def handle_call_tool( name: str, arguments: dict[str, str] | None ) - list[types.TextContent]: # tools is a dictionary with tool names as keys if name not in tools.tools: raise ValueError(fUnknown tool: {name}) tool tools.tools[name] result default try: # invoke the tool result await toolhandler except Exception as e: raise ValueError(fError calling tool {name}: {str(e)}) return [ types.TextContent(typetext, textstr(result)) ]其中要点工具名已作为入参name传入参数则以arguments字典形式传入通过result await toolhandler调用工具参数校验发生在handler属性所指向的函数内部校验失败会抛出异常。到此我们完整掌握了如何用低层服务器实现工具的列出与调用。仓库中的完整可运行示例原文档指向的 code 目录 中提供了 Python 与 TypeScript 两个可运行实现与上文思路一一对应。Python 示例服务器入口 server.py 使用mcp.server.lowlevel.Server创建低层服务器server.list_tools()遍历tools字典将每个工具转换为types.Tool其中convert_to_json(model_cls)把 Pydantic 模型的schema()输出转换为 MCP 需要的 JSON Schema提取properties与required字段server.call_tool()按名称查字典、调用 handler并把异常包装为ValueError返回给客户端run()通过mcp.server.stdio.stdio_server()建立 stdio 双向流并以InitializationOptions声明服务器名example-server、版本0.1.0与能力列表。工具目录结构为 tools/__init__.py聚合add工具add.py 用 Pydantic 校验并计算结果schema.py 定义AddInputModel。客户端 client.py 演示了完整调用链通过stdio_client建立连接 →ClientSession.initialize()→session.list_tools()列出工具 →session.call_tool(add, {a: 5, b: 3})调用工具。运行方式详见 python/README.mdpython -m venv venv source ./venv/bin/activate pip install mcp[cli] python client.py预期输出Available tools: [add] Result of add tool: metaNone content[TextContent(typetext, text8.0, annotationsNone, metaNone)] structuredContentNone isErrorFalseTypeScript 示例服务器入口 server.ts 使用modelcontextprotocol/sdk的Server并显式声明capabilities: { tools: {} }两个请求处理器与上文一致列表处理器直接返回tools数组调用处理器先用tool.rawSchemaZod解析参数成功则执行tool.callback失败返回invalid_arguments工具不存在返回tool_not_found。工具注册在 tools/index.ts 中通过tools.push(addTool)/tools.push(subtractTool)完成——从源码注释可见这些 push 操作替代了普通服务器中的server.tool(...)注册调用。运行与测试详见 typescript/README.mdnpm install npm run build npm start使用 MCP Inspector 验证能力启动 Web 界面npx modelcontextprotocol/inspector node build/app.jsCLI 模式下列出工具npx modelcontextprotocol/inspector --cli node ./build/app.js --method tools/list输出会包含add、subtract两个工具及其由zodToJsonSchema生成的 JSON Schematype: object、properties.a/b: number、required: [a,b]。调用工具npx modelcontextprotocol/inspector --cli node ./build/app.js --method tools/call --tool-name add --tool-arg a1 --tool-arg b2调用不存在的工具add2会返回tool_not_found传入 schema 不允许的参数c例如--tool-arg a1 --tool-arg c2会返回invalid_arguments并附上 Zod 的详细校验错误expected: number、received: undefined、path: [b]。这些输出正是校验机制生效的直接证据。作业扩展与反思基于给出的代码继续扩展添加若干工具、资源和提示词并思考你会发现只需在tools目录中新增文件而无需改动服务器主文件——这正是低层服务器架构带来的扩展红利。本练习不提供参考答案No solution given建议独立完成后再对照自己的实现复盘。小结本章我们看到了低层服务器的工作方式以及它如何帮助我们创建整洁、可持续构建的架构。我们还讨论了校验并演示了如何借助校验库Pydantic / Zod为输入校验创建 schema。下一步下一篇Simple Authentication——在低层服务器基础上加入简单身份认证。赞分享教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载相关推荐aiohttp 低层服务器Low Level Server完全指南使用 web.Server 与 Runner/Site 构建无路由 HTTP 服务aiohttp 低层服务器Low Level Server完全指南使用 web.Server 与 Runner/Site 构建无路由 HTTP 服务 本篇后端Web框架WebSocketMCP Python SDK 进阶篇低层 Server、分页、中间件、扩展与 MCP Apps 完全指南MCP Python SDK 进阶篇低层 Server、分页、中间件、扩展与 MCP Apps 完全指南 在 Model Context ProtocolM人工智能MCP 服务MCP Clients使用 Rube MCP 在 Codex 中自动化 ApilioComposio 技能实战指南使用 Rube MCP 在 Codex 中自动化 ApilioComposio 技能实战指南 本篇技术指南以 composio skills/apilio a教程文档人工智能上一篇Speech-to-Speech 项目 OpenAI Realtime 引擎全解析WebSocket/WebRTC 架构、工具调用与打断处理实战指南下一篇Effect SQL 迁移器并发锁处理解析仅将重复迁移行插入视为并发迁移锁创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表