27 KiB
Node/TypeScript MCP 服务端实现指南
概述
本文档提供了使用 MCP TypeScript SDK 实现 MCP 服务端时的 Node/TypeScript 特定最佳实践与示例。内容涵盖项目结构、服务端初始化、工具注册模式、基于 Zod 的输入验证、错误处理以及完整的可运行示例。
快速参考
关键导入
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";
服务端初始化
const server = new McpServer({
name: "service-mcp-server",
version: "1.0.0"
});
工具注册模式
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)- 所有参数和返回值类型需显式声明
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<typeof UserSearchInputSchema>;
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 状态码)
- 搜索结果为空时返回 "未找到与 '<query>' 匹配的用户"`,
inputSchema: UserSearchInputSchema,
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
openWorldHint: true
}
},
async (params: UserSearchInput) => {
try {
// 输入验证由 Zod schema 处理
// 使用验证后的参数发起 API 请求
const data = await makeApiRequest<any>(
"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 提供运行时类型验证:
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("跳过的结果数")
});
响应格式选项
支持多种输出格式以提高灵活性:
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 格式:
- 返回适合程序化处理的完整结构化数据
- 包含所有可用字段和元数据
- 使用一致的字段名和类型
分页实现
适用于列出资源的工具:
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<typeof ListSchema>) {
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 常量以防止响应过长:
// 在 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;
}
错误处理
提供清晰、可操作的错误信息:
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)}`;
}
共享工具函数
将通用功能提取为可复用的函数:
// 共享 API 请求函数
async function makeApiRequest<T>(
endpoint: string,
method: "GET" | "POST" | "PUT" | "DELETE" = "GET",
data?: any,
params?: any
): Promise<T> {
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:
// 好的做法:异步网络请求
async function fetchData(resourceId: string): Promise<ResourceData> {
const response = await axios.get(`${API_URL}/resource/${resourceId}`);
return response.data;
}
// 不好的做法:Promise 链
function fetchData(resourceId: string): Promise<ResourceData> {
return axios.get(`${API_URL}/resource/${resourceId}`)
.then(response => response.data); // 可读性和可维护性较差
}
TypeScript 最佳实践
- 使用严格 TypeScript:在 tsconfig.json 中启用严格模式
- 定义接口:为所有数据结构创建清晰的接口定义
- 避免使用
any:使用正确类型或unknown替代any - 使用 Zod 进行运行时验证:使用 Zod schema 验证外部数据
- 类型守卫:为复杂类型检查创建类型守卫函数
- 错误处理:始终使用 try-catch 并配合正确的错误类型检查
- 空值安全:使用可选链(
?.)和空值合并(??)
// 好的做法:通过 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<typeof UserSchema>;
async function getUser(id: string): Promise<User> {
const data = await apiCall(`/users/${id}`);
return UserSchema.parse(data); // 运行时验证
}
// 不好的做法:使用 any
async function getUser(id: string): Promise<any> {
return await apiCall(`/users/${id}`); // 没有类型安全
}
包配置
package.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
{
"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"]
}
完整示例
#!/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<typeof UserSearchInputSchema>;
// 共享工具函数
async function makeApiRequest<T>(
endpoint: string,
method: "GET" | "POST" | "PUT" | "DELETE" = "GET",
data?: any,
params?: any
): Promise<T> {
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 的高效访问:
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(推荐用于远程服务端)
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(用于本地集成)
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const transport = new StdioServerTransport();
await server.connect(transport);
传输方式选择:
- 流式 HTTP:Web 服务、远程访问、多客户端
- stdio:命令行工具、本地开发、子进程集成
通知支持
在服务端状态变化时通知客户端:
// 工具列表变更时通知
server.notification({
method: "notifications/tools/list_changed"
});
// 资源变更时通知
server.notification({
method: "notifications/resources/list_changed"
});
适度使用通知——仅当服务端能力确实发生变化时使用。
代码最佳实践
代码组合性与可复用性
你的实现必须优先考虑组合性和代码复用:
-
提取通用功能:
- 为多个工具中使用的操作创建可复用的辅助函数
- 构建共享的 API 客户端来处理 HTTP 请求,而非重复代码
- 将错误处理逻辑集中到工具函数中
- 将业务逻辑提取到可组合的专用函数中
- 提取共享的 Markdown 或 JSON 字段选择与格式化功能
-
避免重复:
- 切勿在不同工具之间复制粘贴相似代码
- 如果发现自己两次编写相似逻辑,将其提取为函数
- 分页、过滤、字段选择和格式化等常见操作应共享
- 认证/授权逻辑应集中管理
构建与运行
始终先构建 TypeScript 代码再运行:
# 构建项目
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<T>返回类型 - 错误处理使用正确的类型守卫(例如
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 - 所有导入正确解析
- 示例工具调用按预期工作