# Node/TypeScript MCP 服务端实现指南 ## 概述 本文档提供了使用 MCP TypeScript SDK 实现 MCP 服务端时的 Node/TypeScript 特定最佳实践与示例。内容涵盖项目结构、服务端初始化、工具注册模式、基于 Zod 的输入验证、错误处理以及完整的可运行示例。 --- ## 快速参考 ### 关键导入 ```typescript import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import express from "express"; import { z } from "zod"; ``` ### 服务端初始化 ```typescript const server = new McpServer({ name: "service-mcp-server", version: "1.0.0" }); ``` ### 工具注册模式 ```typescript server.registerTool( "tool_name", { title: "Tool Display Name", description: "What the tool does", inputSchema: { param: z.string() }, outputSchema: { result: z.string() } }, async ({ param }) => { const output = { result: `Processed: ${param}` }; return { content: [{ type: "text", text: JSON.stringify(output) }], structuredContent: output // 结构化数据的现代模式 }; } ); ``` --- ## MCP TypeScript SDK 官方 MCP TypeScript SDK 提供: - `McpServer` 类用于服务端初始化 - `registerTool` 方法用于工具注册 - Zod schema 集成实现运行时输入验证 - 类型安全的工具处理函数实现 **重要——仅使用现代 API:** - **应该使用**:`server.registerTool()`、`server.registerResource()`、`server.registerPrompt()` - **不要使用**:旧的已弃用 API,如 `server.tool()`、`server.setRequestHandler(ListToolsRequestSchema, ...)` 或手动注册处理函数 - `register*` 方法提供更好的类型安全性、自动 schema 处理,是推荐做法 详情请参阅参考资料中的 MCP SDK 文档。 ## 服务端命名规范 Node/TypeScript MCP 服务端必须遵循以下命名模式: - **格式**:`{service}-mcp-server`(小写,用连字符连接) - **示例**:`github-mcp-server`、`jira-mcp-server`、`stripe-mcp-server` 名称应: - 通用(不与特定功能绑定) - 描述所集成的服务/API - 易于从任务描述中推断 - 不包含版本号或日期 ## 项目结构 为 Node/TypeScript MCP 服务端创建以下结构: ``` {service}-mcp-server/ ├── package.json ├── tsconfig.json ├── README.md ├── src/ │ ├── index.ts # 主入口文件,包含 McpServer 初始化 │ ├── types.ts # TypeScript 类型定义与接口 │ ├── tools/ # 工具实现(每个领域一个文件) │ ├── services/ # API 客户端与共享工具函数 │ ├── schemas/ # Zod 验证 schema │ └── constants.ts # 共享常量(API_URL、CHARACTER_LIMIT 等) └── dist/ # 构建后的 JavaScript 文件(入口:dist/index.js) ``` ## 工具实现 ### 工具命名 工具名称使用蛇形命名法(snake_case),例如 "search_users"、"create_project"、"get_channel_info",名称应清晰且面向操作。 **避免命名冲突**:包含服务上下文以防止重叠: - 使用 "slack_send_message" 而非 "send_message" - 使用 "github_create_issue" 而非 "create_issue" - 使用 "asana_list_tasks" 而非 "list_tasks" ### 工具结构 工具通过 `registerTool` 方法注册,需满足以下要求: - 使用 Zod schema 进行运行时输入验证和类型安全 - `description` 字段必须显式提供——JSDoc 注释不会被自动提取 - 显式提供 `title`、`description`、`inputSchema` 和 `annotations` - `inputSchema` 必须是 Zod schema 对象(而非 JSON schema) - 所有参数和返回值类型需显式声明 ```typescript import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { z } from "zod"; const server = new McpServer({ name: "example-mcp", version: "1.0.0" }); // 用于输入验证的 Zod schema const UserSearchInputSchema = z.object({ query: z.string() .min(2, "查询字符串至少需要 2 个字符") .max(200, "查询字符串不能超过 200 个字符") .describe("用于匹配姓名/邮箱的搜索字符串"), limit: z.number() .int() .min(1) .max(100) .default(20) .describe("返回的最大结果数"), offset: z.number() .int() .min(0) .default(0) .describe("分页时跳过的结果数"), response_format: z.nativeEnum(ResponseFormat) .default(ResponseFormat.MARKDOWN) .describe("输出格式:'markdown' 用于人类可读,'json' 用于机器可读") }).strict(); // 从 Zod schema 推导的类型定义 type UserSearchInput = z.infer; server.registerTool( "example_search_users", { title: "Search Example Users", description: `在 Example 系统中按姓名、邮箱或团队搜索用户。 该工具搜索 Example 平台中的所有用户资料,支持部分匹配和多种搜索筛选条件。不会创建或修改用户,仅搜索已有用户。 参数: - query (string):用于匹配姓名/邮箱的搜索字符串 - limit (number):返回的最大结果数,范围 1-100(默认值:20) - offset (number):分页时跳过的结果数(默认值:0) - response_format ('markdown' | 'json'):输出格式(默认值:'markdown') 返回: JSON 格式:结构化数据,schema 如下: { "total": number, // 找到的匹配总数 "count": number, // 本次响应中的结果数 "offset": number, // 当前分页偏移量 "users": [ { "id": string, // 用户 ID(例如 "U123456789") "name": string, // 全名(例如 "张三") "email": string, // 邮箱地址 "team": string, // 团队名称(可选) "active": boolean // 用户是否活跃 } ], "has_more": boolean, // 是否还有更多结果 "next_offset": number // 下一页的偏移量(若 has_more 为 true) } 示例: - 何时使用:"查找所有营销团队成员" -> 参数为 query="team:marketing" - 何时使用:"搜索张三的账号" -> 参数为 query="zhang" - 不要使用:需要创建用户时(应使用 example_create_user) 错误处理: - 请求过多时返回 "Error: Rate limit exceeded"(429 状态码) - 搜索结果为空时返回 "未找到与 '' 匹配的用户"`, inputSchema: UserSearchInputSchema, annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: true } }, async (params: UserSearchInput) => { try { // 输入验证由 Zod schema 处理 // 使用验证后的参数发起 API 请求 const data = await makeApiRequest( "users/search", "GET", undefined, { q: params.query, limit: params.limit, offset: params.offset } ); const users = data.users || []; const total = data.total || 0; if (!users.length) { return { content: [{ type: "text", text: `未找到与 '${params.query}' 匹配的用户` }] }; } // 准备结构化输出 const output = { total, count: users.length, offset: params.offset, users: users.map((user: any) => ({ id: user.id, name: user.name, email: user.email, ...(user.team ? { team: user.team } : {}), active: user.active ?? true })), has_more: total > params.offset + users.length, ...(total > params.offset + users.length ? { next_offset: params.offset + users.length } : {}) }; // 根据请求格式生成文本表示 let textContent: string; if (params.response_format === ResponseFormat.MARKDOWN) { const lines = [`# 用户搜索结果:'${params.query}'`, "", `共找到 ${total} 个用户(显示 ${users.length} 个)`, ""]; for (const user of users) { lines.push(`## ${user.name}(${user.id})`); lines.push(`- **邮箱**:${user.email}`); if (user.team) lines.push(`- **团队**:${user.team}`); lines.push(""); } textContent = lines.join("\n"); } else { textContent = JSON.stringify(output, null, 2); } return { content: [{ type: "text", text: textContent }], structuredContent: output // 结构化数据的现代模式 }; } catch (error) { return { content: [{ type: "text", text: handleApiError(error) }] }; } } ); ``` ## Zod Schema 用于输入验证 Zod 提供运行时类型验证: ```typescript import { z } from "zod"; // 带验证的基本 schema const CreateUserSchema = z.object({ name: z.string() .min(1, "姓名为必填项") .max(100, "姓名不能超过 100 个字符"), email: z.string() .email("邮箱格式无效"), age: z.number() .int("年龄必须为整数") .min(0, "年龄不能为负数") .max(150, "年龄不能超过 150") }).strict(); // 使用 .strict() 禁止额外字段 // 枚举 enum ResponseFormat { MARKDOWN = "markdown", JSON = "json" } const SearchSchema = z.object({ response_format: z.nativeEnum(ResponseFormat) .default(ResponseFormat.MARKDOWN) .describe("输出格式") }); // 带默认值的可选字段 const PaginationSchema = z.object({ limit: z.number() .int() .min(1) .max(100) .default(20) .describe("返回的最大结果数"), offset: z.number() .int() .min(0) .default(0) .describe("跳过的结果数") }); ``` ## 响应格式选项 支持多种输出格式以提高灵活性: ```typescript enum ResponseFormat { MARKDOWN = "markdown", JSON = "json" } const inputSchema = z.object({ query: z.string(), response_format: z.nativeEnum(ResponseFormat) .default(ResponseFormat.MARKDOWN) .describe("输出格式:'markdown' 用于人类可读,'json' 用于机器可读") }); ``` **Markdown 格式**: - 使用标题、列表和格式化以提高可读性 - 将时间戳转换为人类可读格式 - 显示名称及括号中的 ID - 省略冗长的元数据 - 按逻辑对相关信息分组 **JSON 格式**: - 返回适合程序化处理的完整结构化数据 - 包含所有可用字段和元数据 - 使用一致的字段名和类型 ## 分页实现 适用于列出资源的工具: ```typescript const ListSchema = z.object({ limit: z.number().int().min(1).max(100).default(20), offset: z.number().int().min(0).default(0) }); async function listItems(params: z.infer) { const data = await apiRequest(params.limit, params.offset); const response = { total: data.total, count: data.items.length, offset: params.offset, items: data.items, has_more: data.total > params.offset + data.items.length, next_offset: data.total > params.offset + data.items.length ? params.offset + data.items.length : undefined }; return JSON.stringify(response, null, 2); } ``` ## 字符限制与截断 添加 CHARACTER_LIMIT 常量以防止响应过长: ```typescript // 在 constants.ts 模块级别 export const CHARACTER_LIMIT = 25000; // 最大响应大小(字符数) async function searchTool(params: SearchInput) { let result = generateResponse(data); // 检查字符限制,必要时截断 if (result.length > CHARACTER_LIMIT) { const truncatedData = data.slice(0, Math.max(1, data.length / 2)); response.data = truncatedData; response.truncated = true; response.truncation_message = `响应已从 ${data.length} 项截断至 ${truncatedData.length} 项。` + `请使用 'offset' 参数或添加筛选条件以查看更多结果。`; result = JSON.stringify(response, null, 2); } return result; } ``` ## 错误处理 提供清晰、可操作的错误信息: ```typescript import axios, { AxiosError } from "axios"; function handleApiError(error: unknown): string { if (error instanceof AxiosError) { if (error.response) { switch (error.response.status) { case 404: return "错误:未找到资源。请检查 ID 是否正确。"; case 403: return "错误:权限不足。您没有访问此资源的权限。"; case 429: return "错误:请求频率超限。请稍后再发起更多请求。"; default: return `错误:API 请求失败,状态码 ${error.response.status}`; } } else if (error.code === "ECONNABORTED") { return "错误:请求超时。请重试。"; } } return `错误:发生意外错误:${error instanceof Error ? error.message : String(error)}`; } ``` ## 共享工具函数 将通用功能提取为可复用的函数: ```typescript // 共享 API 请求函数 async function makeApiRequest( endpoint: string, method: "GET" | "POST" | "PUT" | "DELETE" = "GET", data?: any, params?: any ): Promise { try { const response = await axios({ method, url: `${API_BASE_URL}/${endpoint}`, data, params, timeout: 30000, headers: { "Content-Type": "application/json", "Accept": "application/json" } }); return response.data; } catch (error) { throw error; } } ``` ## Async/Await 最佳实践 网络请求和 I/O 操作始终使用 async/await: ```typescript // 好的做法:异步网络请求 async function fetchData(resourceId: string): Promise { const response = await axios.get(`${API_URL}/resource/${resourceId}`); return response.data; } // 不好的做法:Promise 链 function fetchData(resourceId: string): Promise { return axios.get(`${API_URL}/resource/${resourceId}`) .then(response => response.data); // 可读性和可维护性较差 } ``` ## TypeScript 最佳实践 1. **使用严格 TypeScript**:在 tsconfig.json 中启用严格模式 2. **定义接口**:为所有数据结构创建清晰的接口定义 3. **避免使用 `any`**:使用正确类型或 `unknown` 替代 `any` 4. **使用 Zod 进行运行时验证**:使用 Zod schema 验证外部数据 5. **类型守卫**:为复杂类型检查创建类型守卫函数 6. **错误处理**:始终使用 try-catch 并配合正确的错误类型检查 7. **空值安全**:使用可选链(`?.`)和空值合并(`??`) ```typescript // 好的做法:通过 Zod 和接口实现类型安全 interface UserResponse { id: string; name: string; email: string; team?: string; active: boolean; } const UserSchema = z.object({ id: z.string(), name: z.string(), email: z.string().email(), team: z.string().optional(), active: z.boolean() }); type User = z.infer; async function getUser(id: string): Promise { const data = await apiCall(`/users/${id}`); return UserSchema.parse(data); // 运行时验证 } // 不好的做法:使用 any async function getUser(id: string): Promise { return await apiCall(`/users/${id}`); // 没有类型安全 } ``` ## 包配置 ### package.json ```json { "name": "{service}-mcp-server", "version": "1.0.0", "description": "用于 {Service} API 集成的 MCP 服务端", "type": "module", "main": "dist/index.js", "scripts": { "start": "node dist/index.js", "dev": "tsx watch src/index.ts", "build": "tsc", "clean": "rm -rf dist" }, "engines": { "node": ">=18" }, "dependencies": { "@modelcontextprotocol/sdk": "^1.6.1", "axios": "^1.7.9", "zod": "^3.23.8" }, "devDependencies": { "@types/node": "^22.10.0", "tsx": "^4.19.2", "typescript": "^5.7.2" } } ``` ### tsconfig.json ```json { "compilerOptions": { "target": "ES2022", "module": "Node16", "moduleResolution": "Node16", "lib": ["ES2022"], "outDir": "./dist", "rootDir": "./src", "strict": true, "esModuleInterop": true, "skipLibCheck": true, "forceConsistentCasingInFileNames": true, "declaration": true, "declarationMap": true, "sourceMap": true, "allowSyntheticDefaultImports": true }, "include": ["src/**/*"], "exclude": ["node_modules", "dist"] } ``` ## 完整示例 ```typescript #!/usr/bin/env node /** * Example 服务的 MCP 服务端。 * * 本服务端提供与 Example API 交互的工具,包括用户搜索、 * 项目管理和数据导出功能。 */ import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { z } from "zod"; import axios, { AxiosError } from "axios"; // 常量 const API_BASE_URL = "https://api.example.com/v1"; const CHARACTER_LIMIT = 25000; // 枚举 enum ResponseFormat { MARKDOWN = "markdown", JSON = "json" } // Zod schema const UserSearchInputSchema = z.object({ query: z.string() .min(2, "查询字符串至少需要 2 个字符") .max(200, "查询字符串不能超过 200 个字符") .describe("用于匹配姓名/邮箱的搜索字符串"), limit: z.number() .int() .min(1) .max(100) .default(20) .describe("返回的最大结果数"), offset: z.number() .int() .min(0) .default(0) .describe("分页时跳过的结果数"), response_format: z.nativeEnum(ResponseFormat) .default(ResponseFormat.MARKDOWN) .describe("输出格式:'markdown' 用于人类可读,'json' 用于机器可读") }).strict(); type UserSearchInput = z.infer; // 共享工具函数 async function makeApiRequest( endpoint: string, method: "GET" | "POST" | "PUT" | "DELETE" = "GET", data?: any, params?: any ): Promise { try { const response = await axios({ method, url: `${API_BASE_URL}/${endpoint}`, data, params, timeout: 30000, headers: { "Content-Type": "application/json", "Accept": "application/json" } }); return response.data; } catch (error) { throw error; } } function handleApiError(error: unknown): string { if (error instanceof AxiosError) { if (error.response) { switch (error.response.status) { case 404: return "错误:未找到资源。请检查 ID 是否正确。"; case 403: return "错误:权限不足。您没有访问此资源的权限。"; case 429: return "错误:请求频率超限。请稍后再发起更多请求。"; default: return `错误:API 请求失败,状态码 ${error.response.status}`; } } else if (error.code === "ECONNABORTED") { return "错误:请求超时。请重试。"; } } return `错误:发生意外错误:${error instanceof Error ? error.message : String(error)}`; } // 创建 MCP 服务端实例 const server = new McpServer({ name: "example-mcp", version: "1.0.0" }); // 注册工具 server.registerTool( "example_search_users", { title: "Search Example Users", description: `[完整描述如上所示]`, inputSchema: UserSearchInputSchema, annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: true } }, async (params: UserSearchInput) => { // 实现如上所示 } ); // 主函数 // 用于 stdio(本地): async function runStdio() { if (!process.env.EXAMPLE_API_KEY) { console.error("错误:需要设置 EXAMPLE_API_KEY 环境变量"); process.exit(1); } const transport = new StdioServerTransport(); await server.connect(transport); console.error("MCP 服务端通过 stdio 运行"); } // 用于流式 HTTP(远程): async function runHTTP() { if (!process.env.EXAMPLE_API_KEY) { console.error("错误:需要设置 EXAMPLE_API_KEY 环境变量"); process.exit(1); } const app = express(); app.use(express.json()); app.post('/mcp', async (req, res) => { const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined, enableJsonResponse: true }); res.on('close', () => transport.close()); await server.connect(transport); await transport.handleRequest(req, res, req.body); }); const port = parseInt(process.env.PORT || '3000'); app.listen(port, () => { console.error(`MCP 服务端运行在 http://localhost:${port}/mcp`); }); } // 根据环境选择传输方式 const transport = process.env.TRANSPORT || 'stdio'; if (transport === 'http') { runHTTP().catch(error => { console.error("服务端错误:", error); process.exit(1); }); } else { runStdio().catch(error => { console.error("服务端错误:", error); process.exit(1); }); } ``` --- ## 高级 MCP 功能 ### 资源注册 将数据暴露为资源,实现基于 URI 的高效访问: ```typescript import { ResourceTemplate } from "@modelcontextprotocol/sdk/types.js"; // 使用 URI 模板注册资源 server.registerResource( { uri: "file://documents/{name}", name: "Document Resource", description: "按名称访问文档", mimeType: "text/plain" }, async (uri: string) => { // 从 URI 中提取参数 const match = uri.match(/^file:\/\/documents\/(.+)$/); if (!match) { throw new Error("URI 格式无效"); } const documentName = match[1]; const content = await loadDocument(documentName); return { contents: [{ uri, mimeType: "text/plain", text: content }] }; } ); // 动态列出可用资源 server.registerResourceList(async () => { const documents = await getAvailableDocuments(); return { resources: documents.map(doc => ({ uri: `file://documents/${doc.name}`, name: doc.name, mimeType: "text/plain", description: doc.description })) }; }); ``` **何时使用资源 vs 工具:** - **资源**:适用于基于简单 URI 参数的数据访问 - **工具**:适用于需要验证和业务逻辑的复杂操作 - **资源**:数据相对静态或基于模板时 - **工具**:操作有副作用或涉及复杂工作流时 ### 传输选项 TypeScript SDK 支持两种主要传输机制: #### 流式 HTTP(推荐用于远程服务端) ```typescript import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import express from "express"; const app = express(); app.use(express.json()); app.post('/mcp', async (req, res) => { // 为每个请求创建新的传输(无状态,防止请求 ID 冲突) const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined, enableJsonResponse: true }); res.on('close', () => transport.close()); await server.connect(transport); await transport.handleRequest(req, res, req.body); }); app.listen(3000); ``` #### stdio(用于本地集成) ```typescript import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; const transport = new StdioServerTransport(); await server.connect(transport); ``` **传输方式选择:** - **流式 HTTP**:Web 服务、远程访问、多客户端 - **stdio**:命令行工具、本地开发、子进程集成 ### 通知支持 在服务端状态变化时通知客户端: ```typescript // 工具列表变更时通知 server.notification({ method: "notifications/tools/list_changed" }); // 资源变更时通知 server.notification({ method: "notifications/resources/list_changed" }); ``` 适度使用通知——仅当服务端能力确实发生变化时使用。 --- ## 代码最佳实践 ### 代码组合性与可复用性 你的实现必须优先考虑组合性和代码复用: 1. **提取通用功能**: - 为多个工具中使用的操作创建可复用的辅助函数 - 构建共享的 API 客户端来处理 HTTP 请求,而非重复代码 - 将错误处理逻辑集中到工具函数中 - 将业务逻辑提取到可组合的专用函数中 - 提取共享的 Markdown 或 JSON 字段选择与格式化功能 2. **避免重复**: - 切勿在不同工具之间复制粘贴相似代码 - 如果发现自己两次编写相似逻辑,将其提取为函数 - 分页、过滤、字段选择和格式化等常见操作应共享 - 认证/授权逻辑应集中管理 ## 构建与运行 始终先构建 TypeScript 代码再运行: ```bash # 构建项目 npm run build # 运行服务端 npm start # 开发模式(自动重载) npm run dev ``` 始终确保 `npm run build` 成功完成后再认为实现完成。 ## 质量检查清单 在最终确定 Node/TypeScript MCP 服务端实现之前,请确保: ### 战略设计 - [ ] 工具支持完整工作流,而不仅仅是 API 端点封装 - [ ] 工具名称反映自然的任务划分 - [ ] 响应格式针对智能体上下文效率进行优化 - [ ] 在适当的地方使用人类可读的标识符 - [ ] 错误信息引导智能体正确使用 ### 实现质量 - [ ] 聚焦实现:实现最重要和最有价值的工具 - [ ] 所有工具使用 `registerTool` 注册,并带有完整配置 - [ ] 所有工具包含 `title`、`description`、`inputSchema` 和 `annotations` - [ ] 注解正确设置(readOnlyHint、destructiveHint、idempotentHint、openWorldHint) - [ ] 所有工具使用 Zod schema 进行运行时输入验证,并启用 `.strict()` 约束 - [ ] 所有 Zod schema 具有适当的约束条件和描述性错误信息 - [ ] 所有工具具有全面的描述,包含显式的输入/输出类型 - [ ] 描述包含返回值示例和完整的 schema 文档 - [ ] 错误信息清晰、可操作且具有教育意义 ### TypeScript 质量 - [ ] 为所有数据结构定义 TypeScript 接口 - [ ] 在 tsconfig.json 中启用严格 TypeScript 模式 - [ ] 不使用 `any` 类型——使用 `unknown` 或正确的类型替代 - [ ] 所有异步函数具有显式的 `Promise` 返回类型 - [ ] 错误处理使用正确的类型守卫(例如 `axios.isAxiosError`、`z.ZodError`) ### 高级功能(如适用) - [ ] 为适当的数据端点注册资源 - [ ] 配置了合适的传输方式(stdio 或流式 HTTP) - [ ] 为动态服务端能力实现通知 - [ ] 通过 SDK 接口实现类型安全 ### 项目配置 - [ ] Package.json 包含所有必要的依赖 - [ ] 构建脚本在 dist/ 目录中生成可运行的 JavaScript - [ ] 主入口正确配置为 dist/index.js - [ ] 服务端名称遵循格式:`{service}-mcp-server` - [ ] tsconfig.json 正确配置,启用严格模式 ### 代码质量 - [ ] 分页已正确实现(如适用) - [ ] 大型响应检查 CHARACTER_LIMIT 常量并附带清晰信息截断 - [ ] 为可能较大的结果集提供过滤选项 - [ ] 所有网络操作优雅处理超时和连接错误 - [ ] 通用功能被提取为可复用的函数 - [ ] 相似操作的返回类型保持一致 ### 测试与构建 - [ ] `npm run build` 成功完成且无错误 - [ ] dist/index.js 已创建且可执行 - [ ] 服务端可运行:`node dist/index.js --help` - [ ] 所有导入正确解析 - [ ] 示例工具调用按预期工作