# MCP（模型上下文协议）- AI 原生服务器开发 ## 概述模型上下文协议 (MCP) 是一种用于将 AI 助手连接到外部数据源和工具的开放标准。构建公开**工具**（LLM 可以调用的函数）、**资源**（LLM 可以读取的数据）和**提示**（LLM 可以使用的模板）的服务器。 **关键概念**： * **工具**：LLM 可以执行的功能（读取文件、查询 API、运行命令） * **资源** ：法学硕士可以访问的数据源（文件、数据库、API） * **提示**：可重用模板，带有常见任务的参数 * **客户端-服务器** ：MCP 服务器公开功能，客户端（如 Claude Desktop）使用它们 * **传输** ：STDIO（本地）、SSE（服务器发送事件）、HTTP（网络） **官方 SDK** ： * TypeScript：`@modelcontextprotocol/sdk` * Python：`mcp` **安装**： # TypeScript 服务器 npx @modelcontextprotocol/create-server@latest my-server cd my-server && npm install # Python 服务器 pip 安装 mcp # 或者使用 uv （推荐） uv pip 安装 mcp ## 快速入门 - TypeScript 服务器 ### 1\.使用 CLI 创建服务器 # 交互设置 npx @modelcontextprotocol/create-server@latest my-filesystem-server # 选项提示： # - 服务器名称：my-filesystem-server # - 语言：TypeScript # - 包括示例工具：是 ### 2\.定义工具 // src/index.ts 从“@modelcontextprotocol/sdk/server/index.js”导入{服务器}；从“@modelcontextprotocol/sdk/server/stdio.js”导入{ StdioServerTransport }；导入{ 调用ToolRequestSchema，列表工具请求架构，来自“@modelcontextprotocol/sdk/types.js”；从“fs/promises”导入*作为fs；从“路径”导入*作为路径；常量服务器 = 新服务器（ { 名称：“文件系统服务器”，版本：“1.0.0”， }, { 能力：{ 工具：{}， }, } ）； // 列出可用的工具 server.setRequestHandler(ListToolsRequestSchema, async () => { 返回{ 工具：[ { 名称：“读取文件”，描述：“读取文件内容”，输入架构：{ 类型：“对象”，属性：{ 路径：{ 类型：“字符串”，描述：“要读取的文件的路径”， }, }, 必需：[“路径”]， }, }, { 名称：“列表目录”，描述：“列出目录的内容”，输入架构：{ 类型：“对象”，属性：{ 路径：{ 类型：“字符串”，描述：“列表的目录路径”， }, }, 必需：[“路径”]， }, }, ], }; }); // 处理工具调用 server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { const { 名称，参数：args } = request.params; 开关（名称）{ 案例“读取文件”：{ const filePath = args.path 作为字符串； const content = wait fs.readFile(filePath, "utf-8"); 返回{ 内容：[{类型：“文本”，文本：内容}]， }; } 案例“列表目录”：{ const dirPath = args.path 作为字符串； const Entry = wait fs.readdir(dirPath, { withFileTypes: true }); const 列表 = 条目 .map((entry) => `${entry.isDirectory() ? "📁" : "📄"} ${entry.name}`) .join("\n"); 返回{ 内容：[{类型：“文本”，文本：列表}]， }; } 默认： throw new Error(`未知工具：${name}`); } }); // 启动服务器异步函数 main() { const 传输 = new StdioServerTransport(); 等待服务器.connect(传输); console.error("在 stdio 上运行的文件系统 MCP 服务器"); } 主要的（）; ### 3\.配置克劳德桌面 // ~/Library/Application Support/Claude/claude_desktop_config.json (macOS) // %APPDATA%/Claude/claude_desktop_config.json (Windows) { “mcp服务器”：{ “文件系统”：{ “命令”：“节点”， “args”：[“/绝对/路径/到/my-filesystem-server/build/index.js”] } } } ### 4\.构建和测试 # 构建 TypeScript npm 运行构建 # 重新启动 Claude 桌面（Cmd+Q 并重新打开） # 服务器出现在🔌菜单中 ## 快速入门 - Python 服务器 ### 1\.创建服务器 # 服务器.py 导入异步从 mcp.server 导入服务器从 mcp.server.stdio 导入 stdio_server 从 mcp.types 导入工具、文本内容导入 json 导入操作系统 # 创建服务器实例 app = Server(“文件系统服务器”) # 定义工具 @app.list_tools() async def list_tools() -> list[工具]: 返回[ 工具( 名称=“读取文件”， description="读取文件内容", 输入架构={ “类型”：“对象”， “属性”：{ "path": {"type": "string", "description": "要读取的文件路径"} }, “必需”：[“路径”]， }, ), 工具( 名称=“列表目录”， description="列出目录内容", 输入架构={ “类型”：“对象”， “属性”：{ "path": {"type": "string", "description": "目录路径"} }, “必需”：[“路径”]， }, ), ] # 处理工具调用 @app.call_tool() async def call_tool(名称: str, 参数: dict) -> 列表[TextContent]: 如果名称==“读取文件”：文件路径=参数[“路径”] 将 open(file_path, "r") 作为 f：内容 = f.read() 返回[TextContent（类型=“文本”，文本=内容）] elif 名称 == "list_directory": dir_path = 参数[“路径”] 条目 = os.listdir(dir_path) 列表 = "\n".join( f"{'📁' if os.path.isdir(os.path.join(dir_path, e)) else '📄'} {e}" 对于条目中的 e ）返回[TextContent（类型=“文本”，文本=列表）] 其他：引发 ValueError(f"未知工具：{name}") # 启动服务器异步 def main(): 与 stdio_server() 异步为（read_stream，write_stream）：等待app.run（read_stream，write_stream，app.create_initialization_options（））如果 __name__ == "__main__": asyncio.run（主（）） ### 2\.配置克劳德桌面 { “mcp服务器”：{ “文件系统”：{ “命令”：“蟒蛇”， “args”：[“/绝对/路径/to/server.py”] } } } ### 3\.测试 # 测试服务器独立蟒蛇服务器.py # 重新启动克劳德桌面 # 工具出现在🔌菜单中 ## 资源 - 向法学硕士公开数据资源提供对数据源的只读访问。 ### TypeScript 资源示例导入{ 列出资源请求架构，读取资源请求架构，来自“@modelcontextprotocol/sdk/types.js”； // 列出可用资源 server.setRequestHandler(ListResourcesRequestSchema, async () => { 返回{ 资源：[ { uri: "文件:///docs/readme.md", 名称：“自述文件”，描述：“项目自述文件”， mimeType: "文本/markdown", }, { uri：“文件:///config/settings.json”，名称：“设置”，描述：“应用程序设置”， mimeType: "application/json", }, ], }; }); // 处理资源读取 server.setRequestHandler(ReadResourceRequestSchema, 异步(请求) => { const uri = request.params.uri; if (uri.startsWith("file://")) { const filePath = uri.replace("file://", ""); const content = wait fs.readFile(filePath, "utf-8"); 返回{ 内容：[ { 乌里， mimeType: "文本/纯文本", 文本：内容， }, ], }; } 抛出新错误（`未知资源：$ {uri}`）； }); ### Python 资源示例从 mcp.types 导入资源、ResourceContent @app.list_resources() async def list_resources() -> 列表[资源]: 返回[ 资源（ uri =“文件：///docs/readme.md”，名称=“自述文件”，描述=“项目自述文件”， mimeType="文本/markdown", ), 资源（ uri =“文件：///config/settings.json”，名称=“设置”，描述=“应用程序设置”， mimeType =“应用程序/ json”， ), ] @app.read_resource() 异步 def read_resource(uri: str) -> ResourceContent: 如果 uri.startswith("file://"): file_path = uri.replace("file://", "") 将 open(file_path, "r") 作为 f：内容 = f.read() 返回 ResourceContent(uri=uri, mimeType="text/plain", text=content) 引发 ValueError(f"未知资源：{uri}") ## 提示 - 可重复使用的模板提示是法学硕士可以与参数一起使用的模板。 ### TypeScript 提示示例导入{ 列表提示请求架构，获取提示请求架构，来自“@modelcontextprotocol/sdk/types.js”； // 列出提示 server.setRequestHandler(ListPromptsRequestSchema, async () => { 返回{ 提示：[ { 名称：“代码审查”，描述：“审查代码以获得最佳实践”，参数：[ { 名称：“语言”，描述：“编程语言”，要求：真实， }, { 名称：“代码”，描述：“要审查的代码”，要求：真实， }, ], }, ], }; }); // 处理提示请求 server.setRequestHandler(GetPromptRequestSchema, 异步(请求) => { const { 名称，参数：args } = request.params; if (name === "code_review") { const 语言 = args?.语言 || “未知”； const 代码 = args?.code || “”；返回{ 消息：[ { 角色：“用户”，内容：{ 类型：“文本”， text: `查看此 ${language} 代码以了解最佳实践、安全问题和改进：\n\n\`\`\`${语言}\n${code}\n\`\`\``, }, }, ], }; } throw new Error(`未知提示：${name}`); }); ### Python 提示示例从 mcp.types 导入提示、提示消息、提示参数 @app.list_prompts() 异步 def list_prompts() -> 列表[提示]: 返回[ 提示（名称=“代码审查”， description="查看代码以获得最佳实践", 参数=[ 提示参数( 名称=“语言”，描述=“编程语言”，必需= True ), PromptArgument（名称=“代码”，描述=“要审查的代码”，必需= True）， ], ） ] @app.get_prompt() async def get_prompt(name: str, argument: dict) -> list[PromptMessage]: 如果名称==“代码审查”：语言 = argument.get("语言", "未知") 代码 = argument.get("代码", "") 返回[ 提示信息( 角色=“用户”，内容={ “类型”：“文本”， "text": f"查看此{语言}代码：\n\n```{语言}\n{代码}\n```", }, ） ] raise ValueError(f"未知提示：{name}") ## 高级模式 ### 1\.错误处理 // 打字稿 server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { 尝试{ const { 名称，参数：args } = request.params; 如果（名称===“有风险的操作”）{ // 验证输入 if (!args.required_param) { throw new Error("缺少必需参数：required_param"); } // 通过正确的错误处理来执行操作 const 结果 = 等待 PerformRiskyOperation(args.required_param); 返回{ 内容：[{ 类型：“文本”，文本：JSON.stringify(结果) }]， }; } } 捕获（错误）{ // 将错误返回给 LLM 并提供有用的消息返回{ 内容：[ { 类型：“文本”，文本：`错误：${error instanceof 错误？ error.message : 字符串(错误)}`, }, ], 错误：正确， }; } }); # Python @app.call_tool() async def call_tool(名称: str, 参数: dict) -> 列表[TextContent]: 尝试：如果名称==“risky_operation”：如果参数中没有“required_param”：引发 ValueError("缺少必需参数：required_param") 结果=等待执行_risky_操作（参数[“required_param”]）返回 [TextContent(type="text", text=json.dumps(结果))] 除了异常 e：返回 [TextContent(type="text", text=f"错误: {str(e)}", isError=True)] ### 2\.异步操作 // TypeScript - 异步 API 调用从“axios”导入 axios； server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { const { 名称，参数：args } = request.params; 如果（名称===“fetch_api_data”）{ const url = args.url 作为字符串； const 响应 = 等待 axios.get(url); 返回{ 内容：[ { 类型：“文本”，文本： JSON.stringify(response.data, null, 2), }, ], }; } }); # Python - 异步数据库查询导入aiosqlite @app.call_tool() async def call_tool(名称: str, 参数: dict) -> 列表[TextContent]: 如果名称==“查询数据库”：查询=参数[“查询”] 与 aiosqlite.connect("database.db") 异步作为 db：与 db.execute(query) 异步作为游标：行=等待游标.fetchall() 结果 = json.dumps(行) 返回[TextContent（类型=“文本”，文本=结果）] ### 3\.环境变量和配置 // TypeScript - 从环境加载配置从“dotenv”导入 dotenv； dotenv.config(); const API_KEY = process.env.API_KEY; const BASE_URL = process.env.BASE_URL || “https://api.example.com”； server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { if (request.params.name === "api_call") { const 响应 = 等待获取(`${BASE_URL}/端点`, { headers: { 授权: `Bearer ${API_KEY}` }, }); // ... } }); # Python - 从环境加载配置导入操作系统从 dotenv 导入 load_dotenv 加载_dotenv() API_KEY = os.getenv("API_KEY") BASE_URL = os.getenv("BASE_URL", "https://api.example.com") @app.call_tool() async def call_tool(名称: str, 参数: dict) -> 列表[TextContent]: 如果名称==“api_call”： headers = {"授权": f"承载 {API_KEY}"} 与 aiohttp.ClientSession() 异步作为会话： async 与 session.get(f"{BASE_URL}/endpoint", headers=headers) 分别作为：数据 = 等待 resp.json() 返回 [TextContent(type="text", text=json.dumps(data))] ### 4\.流式响应（大数据） // TypeScript - 流式传输大文件内容 server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { if (request.params.name === "read_large_file") { const filePath = request.params.arguments.path 作为字符串； const Stream = fs.createReadStream(filePath, { 编码: "utf-8" }); 让内容=“”； for wait (const 流块) { 内容+=块； // 可以在未来的 MCP 版本中增量生成块 } 返回{ 内容：[{类型：“文本”，文本：内容}]， }; } }); ### 5\.动态工具注册 // TypeScript - 从配置注册工具接口工具配置{ 名称：字符串；描述：字符串；模式：对象；处理程序：（参数：任意）=> Promise<任意>； } const toolRegistry = new Map<字符串, ToolConfig>(); 函数registerTool（配置：ToolConfig）{ toolRegistry.set(config.name, config); } // 注册自定义工具注册工具({ 名称：“自定义工具”， description: "动态注册工具", 架构：{ 类型：“对象”，属性：{ 输入：{ 类型：“字符串” }， }, }, 处理程序：异步（参数）=> { 返回 { 结果：`已处理：${args.input}` }; }, }); // 动态列出工具 server.setRequestHandler(ListToolsRequestSchema, async () => { 返回{ 工具： Array.from(toolRegistry.values()).map((tool) => ({ name: 工具.名称, 描述：工具.描述，输入架构：工具.架构， }））， }; }); // 动态调用工具 server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { const tool = toolRegistry.get(request.params.name); 如果（！工具）{ throw new Error(`未知工具：${request.params.name}`); } const 结果 = 等待 tool.handler(request.params.arguments); 返回{ 内容：[{ 类型：“文本”，文本：JSON.stringify(结果) }]， }; }); ## 运输类型 ### 1\. STDIO（本地执行） Claude 桌面集成的默认设置。服务器作为子进程运行。 // 克劳德桌面配置 { “mcp服务器”：{ “我的服务器”：{ “命令”：“节点”， “args”：[“/path/to/server/build/index.js”]， “环境”：{ "API_KEY": "你的 API 密钥" } } } } ### 2\. SSE（服务器发送事件）对于使用 HTTP 传输的长时间运行的服务器。 // TypeScript SSE 服务器从“@modelcontextprotocol/sdk/server/index.js”导入{服务器}；从“@modelcontextprotocol/sdk/server/sse.js”导入{SSEServerTransport}；从“快递”进口快递；常量应用程序 = Express(); const 服务器 = 新服务器(/* ... */); app.get("/sse", async (req, res) => { const Transport = new SSEServerTransport("/messages", res); 等待服务器.connect(传输); }); app.post("/messages", async (req, res) => { // 处理传入消息 }); app.listen(3000, () => { console.log("MCP 服务器正在监听 http://localhost:3000"); }); // SSE 的 Claude 桌面配置 { “mcp服务器”：{ “远程服务器”：{ “url”：“http://localhost:3000/sse” } } } ### 3\. HTTP（流式 HTTP）适用于 REST 风格的 MCP 服务器。 # 带有 FastAPI 的 Python HTTP 服务器从 fastapi 导入 FastAPI 从 mcp.server.fastapi 导入 create_fastapi_app 应用程序 = FastAPI() mcp_app = Server("http 服务器") # 像往常一样定义工具/资源/提示 # ... #挂载MCP路由 app.mount("/mcp", create_fastapi_app(mcp_app)) 如果 __name__ == "__main__": 进口uvicorn uvicorn.run（应用程序，主机=“0.0.0.0”，端口= 8000） ## 调试 MCP 服务器 ### 1\.服务器日志 // TypeScript - 日志的 Console.error（STDOUT 用于 MCP 协议） server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { console.error(`工具名为：${request.params.name}`); console.error(`参数：${JSON.stringify(request.params.arguments)}`); // ... }); # Python - 使用日志记录模块导入日志记录日志记录.basicConfig（级别=日志记录.DEBUG，文件名=“mcp-server.log”）记录器=logging.getLogger(__name__) @app.call_tool() async def call_tool(名称: str, 参数: dict) -> 列表[TextContent]: logger.debug(f"工具名为：{name}") logger.debug(f"参数：{arguments}") # ... ### 2\. MCP检验员 # 安装MCP Inspector（基于浏览器的调试工具） npm install -g @modelcontextprotocol/inspector # 在你的服务器上运行检查器 mcp-inspector 节点 /path/to/server/build/index.js # 打开浏览器 http://localhost:6274 # - 手动测试工具 # - 查看请求/响应有效负载 # - 调试 JSON 模式验证 ### 3\.克劳德桌面日志 # macOS tail -f ~/Library/Logs/Claude/mcp*.log # 窗口 # 检查%APPDATA%/克劳德/日志/ ### 4\.测试服务器独立 // TypeScript - 添加测试工具 if (process.argv.includes("--test")) { // 模拟工具调用常量测试请求 = { 参数：{ 名称：“读取文件”，参数：{ 路径：“./test.txt” }， }, }; 服务器 .setRequestHandler(CallToolRequestSchema, 异步(请求) => { // ...你的处理程序 }) .then((处理程序) => 处理程序(testRequest)) .then((结果) => console.log(JSON.stringify(结果, null, 2))) .catch((错误) => console.error(错误)); } 否则{ // 正常服务器启动主要（）； } # 不使用 Claude Desktop 进行测试节点构建/index.js --测试 ## 真实服务器示例 ### 1\. GitHub 集成服务器 // github-server/src/index.ts 从“@octokit/rest”导入{Octokit}； const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN }); server.setRequestHandler(ListToolsRequestSchema, async () => { 返回{ 工具：[ { 名称：“list_repos”，描述：“列出用户/组织的存储库”，输入架构：{ 类型：“对象”，属性：{ 所有者：{ 类型：“字符串”，描述：“用户名或组织名称” }， }, 必需：[“所有者”]， }, }, { 名称：“创建问题”，描述：“创建 GitHub 问题”，输入架构：{ 类型：“对象”，属性：{ 所有者：{类型：“字符串”}，回购：{类型：“字符串”}，标题：{ 类型：“字符串” }，主体：{类型：“字符串”}， }, 必需：[“所有者”，“回购”，“标题”]， }, }, ], }; }); server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { const { 名称，参数：args } = request.params; 如果（名称===“list_repos”）{ const { 数据 } = 等待 octokit.repos.listForUser({ 用户名：args.owner 作为字符串， }); const repos = data.map((r) => ({ 名称：r.name，描述：r.描述，星星：r.stargazers_count， })); 返回{ 内容：[{ 类型：“文本”，文本：JSON.stringify(repos, null, 2) }], }; } 如果（名称===“创建问题”）{ const { 数据 } = 等待 octokit.issues.create({ 所有者：args.owner 作为字符串， repo: args.repo 作为字符串， title: args.title 作为字符串， body: args.body 作为字符串， }); 返回{ 内容：[ { type: "text", text: `创建问题: ${data.html_url}` }, ], }; } }); ### 2\.数据库查询服务器 # 数据库服务器/server.py 导入 asyncpg 导入 json 从 mcp.server 导入服务器从 mcp.types 导入工具、文本内容 app = Server("数据库服务器") 异步 def get_db_pool(): 返回等待 asyncpg.create_pool( 主机=“本地主机”，数据库=“mydb”，用户=“用户”，密码=“密码” ） @app.list_tools() async def list_tools() -> list[工具]: 返回[ 工具( 名称=“查询”， description="执行 SQL 查询（仅限 SELECT）", 输入架构={ “类型”：“对象”， “属性”：{ "sql": {"type": "string", "description": "SQL 查询"}, }, “必需”：[“sql”]， }, ), 工具( 名称=“列表表”， description="列出数据库中的所有表", inputSchema={“类型”：“对象”，“属性”：{}}， ), ] @app.call_tool() async def call_tool(名称: str, 参数: dict) -> 列表[TextContent]: 池 = 等待 get_db_pool() 如果名称==“查询”： sql = 参数["sql"] # 安全性：只允许 SELECT 如果不是 sql.strip().upper().startswith("SELECT")：引发 ValueError("仅允许 SELECT 查询") 与 pool.acquire() 异步作为 conn：行=等待conn.fetch(sql) 结果 = [行中行的字典（行）] 返回 [TextContent(type="text", text=json.dumps(结果, indent=2))] elif 名称 == "list_tables": 与 pool.acquire() 异步作为 conn：表 = 等待 conn.fetch(""" 从 information_schema.tables 中选择表名 WHERE table_schema = '公共' “””）结果 = [row["table_name"] 表中的行] 返回 [TextContent(type="text", text=json.dumps(结果, indent=2))] ### 3\.网页抓取服务器 // scraper-server/src/index.ts 从“cheerio”导入*作为cheerio；从“axios”导入 axios； server.setRequestHandler(ListToolsRequestSchema, async () => { 返回{ 工具：[ { 名称：“scrape_url”，描述：“从 URL 中抓取内容”，输入架构：{ 类型：“对象”，属性：{ url: { 类型: "string", 描述: "要抓取的 URL" }, 选择器：{ 类型：“字符串”，描述：“内容的CSS选择器”， }, }, 必需：[“网址”]， }, }, ], }; }); server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { if (request.params.name === "scrape_url") { const url = request.params.arguments.url 作为字符串； const 选择器 = request.params.arguments.selector 作为字符串； const 响应 = 等待 axios.get(url); const $ = Cheerio.load(response.data); 让内容：字符串；如果（选择器）{ 内容= $(选择器).text(); } 否则{ 内容= $("body").text(); } 返回{ 内容：[ { 类型：“文本”，文本：content.trim(), }, ], }; } }); ## 最佳实践 ### 1\.工具设计 **✅ 做**： * 使用清晰的、描述性的工具名称（“read_file”而不是“rf”） * 提供LLM理解的详细描述 * 定义全面的 JSON 模式和描述 * 尽可能返回结构化数据（JSON） * 通过有用的消息优雅地处理错误 **❌不要**： * 创建过于宽泛的工具（拆分复杂的操作） * 返回大量有效负载（对大型数据集进行分页） * 使用不明确的参数名称 * 假设法学硕士了解您的特定领域术语 ### 2\.安全性 **关键规则**： * 验证所有输入（类型、范围、格式） * 清理文件路径（防止目录遍历） * 对命令/操作使用白名单 * 切勿在响应中暴露敏感凭据 * 对昂贵的操作实施速率限制 * 默认使用只读访问权限 // 示例：路径验证函数 validatePath(inputPath: string): string { const 标准化 = path.normalize(inputPath); const allowed = path.resolve("/safe/directory"); if (!normalized.startsWith(允许)) { throw new Error("允许目录之外的路径"); } 返回标准化； } ### 3\.性能 * 缓存昂贵的操作 * 尽可能流式传输大量响应 * 使用分页进行列表操作 * 设置合理的超时时间 * 为限速API实现请求队列 ### 4\.测试 # 使用MCP Inspector进行手动测试 mcp-inspector 节点 build/index.js # 单元测试工具处理程序 npm测试 # 与 Claude Desktop 的集成测试 # 1. 添加到配置 # 2.重启克劳德 # 3. 在对话中进行测试 ### 5\.文档 * 在代码注释中记录所有工具 * 在描述中提供示例用法 * 在文档中包含错误场景 * 维护服务器更新的变更日志 ## 常见陷阱 ❌ **写入 STDOUT** （破坏 MCP 协议）： // 错误 console.log("调试消息"); // STDOUT 用于 MCP 协议 // 正确 console.error("调试消息"); // 日志的 STDERR ❌ **不处理错误**： // 错误 - 未处理的承诺拒绝 server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { const data = 等待riskyOperation(); // 可以抛出返回 { 内容：[{ 类型：“文本”，文本：数据 }] }； }); // 正确 server.setRequestHandler(CallToolRequestSchema, 异步(请求) => { 尝试{ const data = 等待riskyOperation(); 返回 { 内容：[{ 类型：“文本”，文本：数据 }] }； } 捕获（错误）{ 返回{ 内容：[{ 类型：“文本”，文本：`错误：${error.message}` }]，错误：正确， }; } }); ❌ **阻塞操作**： # 错误 - 同步文件读取会阻塞事件循环 @app.call_tool() async def call_tool(名称: str, 参数: dict): with open("large_file.txt", "r") as f: # 块！内容 = f.read() 返回[TextContent（类型=“文本”，文本=内容）] # 正确 - 异步文件 I/O @app.call_tool() async def call_tool(名称: str, 参数: dict): 与 aiofiles.open("large_file.txt", "r") 异步作为 f：内容 = 等待 f.read() 返回[TextContent（类型=“文本”，文本=内容）] ❌ **架构中缺少必填字段**： // 错误 - 缺少“必填”字段 { 名称：“搜索”，输入架构：{ 类型：“对象”，属性：{ 查询：{ 类型：“字符串” } } // 缺少：必需：[“查询”] } } ## 资源 * **官方文档**： * **TypeScript SDK** ： * **Python SDK**： * **服务器示例**： * **MCP 检查器** ： * **规范**： ## 相关技能构建 MCP 服务器时，请考虑以下补充技能： * **typescript-core** ：TypeScript 类型安全、tsconfig 优化和高级模式 * **asyncio** ：具有 async/await 的 MCP 服务器的 Python 异步模式 * **openrouter** ：用于多模型支持的替代 LLM API 集成 ### 快速 TypeScript MCP 模式（内联以供独立使用） // 使用 TypeScript 的类型安全 MCP 服务器从'@modelcontextprotocol/sdk/server/index.js'导入{服务器}；从 '@modelcontextprotocol/sdk/server/stdio.js' 导入 { StdioServerTransport }；导入{ 调用ToolRequestSchema，列表工具请求架构，来自“@modelcontextprotocol/sdk/types.js”；从 'zod' 导入 { z }； // 定义带有运行时验证的工具模式 const SearchSchema = z.object({ 查询：z.string().min(1).max(500), 限制：z.number().int().min(1).max(100).default(10), }); type SearchArgs = z.infer; 类 TypeSafeMCPServer { 私有服务器：服务器；构造函数（）{ this.server = 新服务器( { name: '类型搜索服务器', 版本：'1.0.0'， }, { 能力：{ 工具：{}， }, } ）； this.setupToolHandlers(); } 私人 setupToolHandlers() { // 列出具有完整类型推断的工具 this.server.setRequestHandler(ListToolsRequestSchema, async () => ({ 工具：[ { 名称：'搜索'，描述：“使用类型安全参数进行搜索”，输入架构：{ 类型：'对象'，属性：{ 查询：{ 类型：'字符串'，最小长度：1，最大长度：500 }， limit: { type: 'number', 最小值: 1, 最大值: 100, 默认: 10 }, }, 必需：['查询']， }, }, ], })); // 类型安全工具执行 this.server.setRequestHandler(CallToolRequestSchema, async (请求) => { if (request.params.name === '搜索') { // 使用 Zod 进行运行时验证 const args = SearchSchema.parse(request.params.arguments); // 类型安全的实现 const 结果 = 等待 this.performSearch(args); 返回{ 内容：[{ 类型：'文本'，文本：JSON.stringify(结果) }]， }; } throw new Error(`未知工具：${request.params.name}`); }); } 私有异步执行搜索（参数：SearchArgs）：Promise> { // 具有完全类型安全的实现 // args.query 是字符串，args.limit 是数字返回[]； } 异步运行（）{ const 传输 = new StdioServerTransport(); 等待 this.server.connect(transport); } } // 启动服务器 const 服务器 = new TypeSafeMCPServer(); 服务器.run(); ### 快速 Python 异步 MCP 模式（内联以供独立使用） # 使用 Python 异步 MCP 服务器导入异步导入日志记录从输入导入任何从 mcp.server 导入服务器从 mcp.server.stdio 导入 stdio_server 从 mcp.types 导入工具、文本内容从 pydantic 导入 BaseModel、Field # Pydantic 的类型安全参数模型类 SearchArgs(BaseModel): 查询：str = Field(..., min_length=1, max_length=500) 限制： int = Field(10, ge=1, le=100) AsyncMCPServer 类： def __init__(自身): self.server = Server("异步搜索服务器") self.setup_handlers() def setup_handlers（自身）： @self.server.list_tools() async def list_tools() -> list[工具]: 返回[ 工具( 名称=“搜索”， description="使用异步处理进行搜索", 输入架构={ “类型”：“对象”， “属性”：{ “查询”：{“类型”：“字符串”，“最小长度”：1，“最大长度”：500}， “限制”：{“类型”：“数字”，“最小值”：1，“最大值”：100，“默认”：10}， }, “必需”：[“查询”]， }, ） ] @self.server.call_tool() async def call_tool(名称: str, 参数: dict[str, Any]) -> list[TextContent]: 如果名称==“搜索”： # 使用 Pydantic 进行验证 args = SearchArgs(**参数) # 异步实现结果=等待 self.perform_search(args) 返回[ 文本内容( 类型=“文本”，文本=str（结果）） ] 引发 ValueError(f"未知工具：{name}") 异步 def Perform_search(self, args: SearchArgs) -> list[dict[str, str]]: """异步搜索实现""" # 模拟异步I/O 等待 asyncio.sleep(0.1) # 使用经过验证的参数（args.query 是 str，args.limit 是 int）返回[ {“title”：f“{args.query}的结果”，“url”：“https://example.com”} ] 异步 def main(): """运行异步 MCP 服务器""" 日志记录.basicConfig（级别=日志记录.INFO）服务器 = AsyncMCPServer() # 对 Claude Desktop 使用 stdio 传输与 stdio_server() 异步为（read_stream，write_stream）：等待服务器.服务器.运行（读流，写流， server.server.create_initialization_options() ）如果 __name__ == "__main__": asyncio.run（主（）） ### 快速多语言 MCP 模式（内联供独立使用） **TypeScript 与 Python 的权衡：** 特色|打字稿 |蟒蛇 ---|---|--- **类型安全** |编译时 + 运行时 (Zod) |仅运行时（Pydantic） **性能** |更快的启动，Node.js 开销 |启动速度较慢，更适合 CPU 任务 **异步支持** |本机异步/等待、事件循环 | asyncio，非常适合 I/O **生态系统** | npm 包、前端工具 |数据科学、ML 库 **最适合** | Web API、实时工具 |数据处理、机器学习集成 **两者的共同模式：** 1. **输入验证** * TypeScript：Zod 模式 * Python：Pydantic 模型 2. **错误处理** * 两者：尝试/捕获特定的错误类型 * MCP响应中返回错误内容 3. **资源管理** * TypeScript：async/await 和 try/finally * Python：异步上下文管理器 4. **测试** * TypeScript：带有模拟传输的 Vitest/Jest * Python：使用 pytest-asyncio 进行 pytest **选择实施语言：** // TypeScript - 最适合： // - 文件系统操作 // - 网页抓取/HTTP 请求 // - JSON/API 操作 // - 实时数据流 // Python - 最适合： // - 数据分析（pandas、numpy） // - 机器学习（scikit-learn、torch） // - 数据库 ETL 操作 // - 科学计算 [如果一起部署，各自的技能中可以使用完整的 TypeScript、Python 异步和 OpenRouter 模式] ## 本地 MCP 集成模式（您的存储库） ### .mcp.json 布局（项目范围） { “mcp服务器”：{ “葛记忆”：{ “类型”：“stdio”， “命令”：“kuzu-内存”， “args”：[“mcp”] }, “mcp-矢量搜索”：{ “类型”：“stdio”， “命令”：“紫外线”， “args”：[“运行”，“mcp矢量搜索”，“mcp”]， “环境”：{ "MCP_ENABLE_FILE_WATCHING": "真" } } } } ### CLI 约定 * 为 stdio 服务器提供 `mcp` 子命令。 * 使用“setup”进行端到端初始化+集成（mcp-向量搜索、kuzu-内存）。 * 使用 `install` / `uninstall` 来定位特定客户端 (mcp-ticketer)。 * 提供`doctor`命令来验证依赖关系。 ### 环境约定 * 文件监视：`MCP_ENABLE_FILE_WATCHING=true` * Kuzu内存：`KUZU_MEMORY_PROJECT_ROOT`，`KUZU_MEMORY_DB` * 票务适配器：`MCP_TICKETER_ADAPTER`、`GITHUB_TOKEN`、`GITHUB_OWNER`、`GITHUB_REPO` ### 常用服务器名称 `kuzu-内存`、`mcp-矢量搜索`、`mcp-ticketer`、`mcp-skillset`、`mcp-browser` ## 总结 * **MCP** 通过工具、资源和提示实现 AI 原生服务器开发 * **工具**是LLM可以调用的函数（读取文件、查询API、运行操作） * **资源**提供只读数据访问（文件、数据库、文档） * **提示**是可重用的模板，其中包含常见任务的参数 * **SDK** 可用于 TypeScript 和 Python，并具有完全异步支持 * **Claude Desktop** 通过 JSON 配置集成（STDIO 传输） * **使用 MCP Inspector、服务器日志和独立测试进行调试** * **安全**关键：验证输入、清理路径、使用白名单 * **最佳实践**：清晰的命名、全面的模式、错误处理、性能优化

MCP

包含技能