MCP服务器框架文档
目录
概述
MCP（模型上下文协议）服务器框架是一个基于TypeScript的抽象层，简化了MCP服务器的创建。它提供了一个强大的基类，可以处理传输配置、工具注册和资源管理，使您能够专注于实现业务逻辑。
主要特点
- 🚀 多种运输方式：支持stdio和HTTP流媒体
- 🛠️ 轻松工具注册：用于注册带有验证的工具的简单API
- 📦 资源管理：对服务资源的内置支持
- 🔒 类型安全：通过Zod模式验证完全支持TypeScript
- 🎯 最小沸腾板：专注于你的逻辑，而不是基础设施
- 🔄 可扩展架构：易于扩展和定制
安装
# Install required dependencies
npm install @modelcontextprotocol/sdk zod
# Or using yarn
yarn add @modelcontextprotocol/sdk zod
# Or using pnpm
pnpm add @modelcontextprotocol/sdk zod项目结构
your-project/
├── src/
│ ├── mcp/
│ │ ├── BaseMCPServer.ts # Base class
│ │ ├── examples/
│ │ │ ├── greetings.mcp.server.ts # Example implementation
│ │ │ └── weather.mcp.server.ts # Example implementation
│ │ └── you-server/
│ │ └── your-custom.mcp.server.ts # Your implementation
│ └── app.ts # Entry point
├── package.json
└── tsconfig.json快速开始
1.基本服务器实现
import { BaseMCPServer, BaseMCPConfig } from "./base/BaseMCPServer";
import type { CallToolResult } from "@modelcontextprotocol/sdk/types";
import z from "zod";
export class HelloWorldMCP extends BaseMCPServer {
protected registerComponents(): void {
// Register a simple tool
this.registerToolHelper(
'say_hello',
{
description: 'Say hello to someone',
inputSchema: z.object({
name: z.string()
}),
outputSchema: z.object({
message: z.string()
})
},
this.sayHello.bind(this)
);
}
private sayHello(args: { name: string }): CallToolResult {
return {
content: [{
type: "text",
text: `Hello, ${args.name}!`
}],
structuredContent: {
message: `Hello, ${args.name}!`
}
};
}
}
// Start the server
const server = new HelloWorldMCP({
name: "HelloWorldMCP",
version: "1.0.0",
transportMode: "stdio"
});
await server.start();核心概念
BaseMCP服务器类
这 BaseMCPServer 是一个抽象类，为所有MCP服务器提供基础。它处理：
- 传输初始化和管理
- 配置验证
- 工具和资源注册助手
- 连接生命周期管理
配置架构
每个MCP服务器都需要一个配置对象：
interface BaseMCPConfig {
name: string; // Server name (2-100 chars)
version: string; // Server version (1-10 chars)
host?: string; // Host for HTTP mode (default: "localhost")
port?: number; // Port for HTTP mode (default: 3000)
transportMode: "stdio" | "streamable-http"; // Transport mode
transport?: Transport; // Optional custom transport instance
}运输方式
何时使用每种运输方式
STDIO（标准输入/输出）
transportMode: "stdio"使用案例：
- ✅ CLI工具和命令行应用程序
- ✅ 本地开发和测试
- ✅ 与shell脚本的简单集成
- ✅ 单用户、单会话应用程序
- ✅ 流程之间基于管道的通信
- ✅ 与系统工具集成
优势：
- 设置简单，无需网络配置
- 本地通信延迟低
- 适用于Unix哲学（管道、重定向）
- 无防火墙或网络安全问题
- 非常适合子流程通信
缺点：
- 无法同时处理多个连接
- 仅限于本地机器通信
- 不适合web应用程序
- 无内置会话管理
示例用例：
# Using stdio MCP server in a pipeline
echo '{"method": "greet", "params": {"name": "Alice"}}' | node your-stdio-server.js可流式HTTP（可流式HTTP）
transportMode: "streamable-http"使用案例：
- ✅ Web应用程序和API
- ✅ 多用户应用程序
- ✅ 远程访问场景
- ✅ 微服务架构
- ✅ 云部署
- ✅ 需要会话管理的应用程序
- ✅ 实时流媒体应用程序
优势：
- 支持多个并发连接
- 可以通过网络远程访问
- 内置会话管理
- 适用于标准HTTP基础架构
- 可以与负载平衡器和代理集成
- 支持流式响应
缺点：
- 需要网络配置
- 延迟高于stdio
- 可能需要防火墙配置
- 更复杂的设置
配置示例：
const server = new YourMCP({
name: "WebMCP",
version: "1.0.0",
host: "0.0.0.0", // Listen on all interfaces
port: 8080,
transportMode: "streamable-http"
});运输方式决策树
Is your application...
│
├─ A CLI tool or script?
│ └─ Use STDIO ✓
│
├─ A web service or API?
│ └─ Use Streamable HTTP ✓
│
├─ Used by multiple users simultaneously?
│ └─ Use Streamable HTTP ✓
│
├─ Part of a microservices architecture?
│ └─ Use Streamable HTTP ✓
│
├─ A local development tool?
│ └─ Use STDIO ✓
│
├─ Integrated with shell scripts?
│ └─ Use STDIO ✓
│
└─ Deployed in the cloud?
└─ Use Streamable HTTP ✓创建您的第一个MCP服务器
分步指南
步骤1：定义服务器类
import { BaseMCPServer, BaseMCPConfig } from "./base/BaseMCPServer";
import z from "zod";
export class MyCustomMCP extends BaseMCPServer {
// Add any custom properties
private apiKey?: string;
constructor(params: BaseMCPConfig & { apiKey?: string }) {
super(params);
this.apiKey = params.apiKey;
}
protected registerComponents(): void {
// This method is called during construction
// Register all your tools and resources here
}
}步骤2：实现registerComponents方法
protected registerComponents(): void {
// Register tools
this.registerToolHelper(
'tool_name',
{
description: 'What this tool does',
inputSchema: z.object({
param1: z.string(),
param2: z.number().optional()
}),
outputSchema: z.object({
result: z.string()
})
},
this.toolHandler.bind(this)
);
// Register resources
this.registerResourceHelper(
'resource_name',
'resource://uri',
{
description: 'What this resource provides',
metadata: { /* optional metadata */ }
},
this.resourceHandler.bind(this)
);
}步骤3：实施工具处理程序
private async toolHandler(args: { param1: string; param2?: number }): Promise {
// Your tool logic here
const result = await this.processData(args);
return {
content: [{
type: "text",
text: `Processed: ${result}`
}],
structuredContent: {
result: result,
metadata: { processed: true }
}
};
}步骤4：启动服务器
async function main() {
const server = new MyCustomMCP({
name: "MyCustomMCP",
version: "1.0.0",
transportMode: "stdio", // or "streamable-http"
apiKey: process.env.API_KEY
});
await server.start();
console.log("Server started successfully!");
}
main().catch(console.error);注册工具
工具注册API
工具是MCP服务器公开功能的主要方式。每个工具都有：
- 唯一名称
- 描述
- 输入模式（验证）
- 输出模式（验证）
- 处理函数
基本工具注册
this.registerToolHelper(
'calculate_sum',
{
description: 'Calculate the sum of two numbers',
inputSchema: z.object({
a: z.number(),
b: z.number()
}),
outputSchema: z.object({
sum: z.number()
})
},
(args) => ({
content: [{
type: "text",
text: `The sum is ${args.a + args.b}`
}],
structuredContent: {
sum: args.a + args.b
}
})
);带异步处理程序的高级工具
this.registerToolHelper(
'fetch_data',
{
description: 'Fetch data from an API',
inputSchema: z.object({
endpoint: z.string().url(),
method: z.enum(['GET', 'POST', 'PUT', 'DELETE']).default('GET'),
body: z.record(z.any()).optional()
}),
outputSchema: z.object({
status: z.number(),
data: z.any()
})
},
async (args) => {
const response = await fetch(args.endpoint, {
method: args.method,
body: args.body ? JSON.stringify(args.body) : undefined,
headers: {
'Content-Type': 'application/json'
}
});
const data = await response.json();
return {
content: [{
type: "text",
text: `Response: ${JSON.stringify(data)}`
}],
structuredContent: {
status: response.status,
data: data
}
};
}
);具有复杂验证功能的工具
this.registerToolHelper(
'process_user',
{
description: 'Process user information',
inputSchema: z.object({
name: z.string().min(2).max(50),
email: z.string().email(),
age: z.number().int().min(0).max(120),
preferences: z.object({
newsletter: z.boolean().default(false),
notifications: z.enum(['all', 'important', 'none']).default('important')
}).optional(),
tags: z.array(z.string()).max(10).optional()
}),
outputSchema: z.object({
userId: z.string(),
status: z.enum(['created', 'updated', 'error'])
})
},
async (args) => {
// Process user data
const userId = await this.createOrUpdateUser(args);
return {
content: [{
type: "text",
text: `User ${args.name} processed successfully`
}],
structuredContent: {
userId,
status: 'created'
}
};
}
);注册资源
资源提供客户端可以访问的静态或动态内容。
基本资源注册
this.registerResourceHelper(
'config.json',
'config://settings',
{
description: 'Application configuration',
metadata: {
version: '1.0.0',
lastUpdated: new Date().toISOString()
}
},
async () => {
const config = await this.loadConfiguration();
return {
contents: [{
text: JSON.stringify(config, null, 2),
uri: 'config.json',
mimeType: 'application/json'
}]
};
}
);具有文件读取功能的资源
this.registerResourceHelper(
'document.pdf',
'file://documents/important.pdf',
{
description: 'Important document',
content: '/path/to/document.pdf'
},
async () => {
const pdfContent = await extractPdfTextFromFile('/path/to/document.pdf');
return {
contents: [{
text: pdfContent,
uri: 'document.txt',
mimeType: 'text/plain'
}]
};
}
);动态资源生成
this.registerResourceHelper(
'status.json',
'status://current',
{
description: 'Current system status',
metadata: {
refreshInterval: 30000 // 30 seconds
}
},
async () => {
const status = {
timestamp: new Date().toISOString(),
health: 'healthy',
uptime: process.uptime(),
memory: process.memoryUsage(),
activeConnections: this.getActiveConnections()
};
return {
contents: [{
text: JSON.stringify(status, null, 2),
uri: 'status.json',
mimeType: 'application/json'
}]
};
}
);高级配置
自定义传输配置
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp";
// Create custom HTTP transport with session management
const customTransport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => `session_${Date.now()}_${Math.random()}`,
cors: {
origin: '*',
credentials: true
}
});
const server = new YourMCP({
name: "CustomMCP",
version: "1.0.0",
transportMode: "streamable-http",
transport: customTransport
});基于环境的配置
const config: BaseMCPConfig = {
name: process.env.MCP_NAME || "DefaultMCP",
version: process.env.MCP_VERSION || "1.0.0",
host: process.env.MCP_HOST || "localhost",
port: parseInt(process.env.MCP_PORT || "3000"),
transportMode: (process.env.MCP_TRANSPORT || "stdio") as "stdio" | "streamable-http"
};
const server = new YourMCP(config);多服务器设置
async function startMultipleServers() {
const servers = [
new GreetingsMCP({
name: "GreetingsMCP",
version: "1.0.0",
port: 3001,
transportMode: "streamable-http"
}),
new WeatherMCP({
name: "WeatherMCP",
version: "1.0.0",
port: 3002,
transportMode: "streamable-http"
}),
new CalculatorMCP({
name: "CalculatorMCP",
version: "1.0.0",
transportMode: "stdio"
})
];
await Promise.all(servers.map(server => server.start()));
console.log("All servers started!");
}最佳实践
1.错误处理
始终在工具中实施正确的错误处理：
private async riskyOperation(args: any): Promise {
try {
const result = await this.performOperation(args);
return {
content: [{
type: "text",
text: `Success: ${result}`
}],
structuredContent: { result }
};
} catch (error) {
console.error('Operation failed:', error);
return {
content: [{
type: "text",
text: `Error: ${error.message}`
}],
structuredContent: {
error: true,
message: error.message
}
};
}
}2.输入验证
使用Zod模式进行全面验证：
const EmailSchema = z.string().email().toLowerCase().trim();
const PhoneSchema = z.string().regex(/^\+?[1-9]\d{1,14}$/);
const DateSchema = z.string().datetime();
inputSchema: z.object({
email: EmailSchema,
phone: PhoneSchema.optional(),
birthDate: DateSchema
})3.记录和监控
实施调试和监控日志记录：
export class MonitoredMCP extends BaseMCPServer {
private logger: Logger;
constructor(params: BaseMCPConfig) {
super(params);
this.logger = new Logger(this.name);
}
protected registerComponents(): void {
this.logger.info('Registering components...');
// Register tools
}
private async toolHandler(args: any): Promise {
const startTime = Date.now();
this.logger.debug('Tool called with args:', args);
try {
const result = await this.process(args);
this.logger.info(`Tool completed in ${Date.now() - startTime}ms`);
return result;
} catch (error) {
this.logger.error('Tool failed:', error);
throw error;
}
}
}4.资源缓存
对频繁访问的资源实施缓存：
export class CachedMCP extends BaseMCPServer {
private cache = new Map();
private cacheTTL = 60000; // 1 minute
private async getCachedResource(key: string, loader: () => Promise): Promise {
const cached = this.cache.get(key);
if (cached && Date.now() - cached.timestamp Promise;
async shutdown(): Promise {
console.log('Shutting down server...');
if (this.cleanup) {
await this.cleanup();
}
// Close connections, save state, etc.
await this.disconnect();
console.log('Server shutdown complete');
}
}
// Handle process signals
process.on('SIGINT', async () => {
await server.shutdown();
process.exit(0);
});api参考
BaseMCP服务器
构造函数
constructor(params: BaseMCPConfig)保护方法
protected abstract registerComponents(): void
protected registerToolHelper(
name: string,
config: ToolConfig,
handler: ToolHandler
): void
protected registerResourceHelper(
name: string,
uri: string,
config: ResourceConfig,
handler: ResourceHandler
): void公共方法
public async connect(): Promise
public async start(): Promise
public getHTTPTransporter(): StreamableHTTPServerTransport | undefined
public getStdioTransporter(): StdioServerTransport | undefined
public getTransporter(): Transport
public getMetadata(): ServerMetadata类型
interface BaseMCPConfig {
name: string;
version: string;
host?: string;
port?: number;
transportMode: "stdio" | "streamable-http";
transport?: Transport;
}
interface CallToolResult {
content: Array;
structuredContent?: any;
}
interface ToolConfig {
description: string;
inputSchema: ZodSchema;
outputSchema: ZodSchema;
}
interface ResourceConfig {
description: string;
content?: string;
metadata?: Record;
}
interface ServerMetadata {
name: string;
version: string;
transportMode: TransportMode;
host: string;
port: number;
}例子
完整数据库查询服务器
import { BaseMCPServer, BaseMCPConfig } from "./base/BaseMCPServer";
import { Database } from 'your-database-library';
import z from "zod";
export class DatabaseMCP extends BaseMCPServer {
private db: Database;
constructor(params: BaseMCPConfig & { databaseUrl: string }) {
super(params);
this.db = new Database(params.databaseUrl);
}
protected registerComponents(): void {
// Query tool
this.registerToolHelper(
'query',
{
description: 'Execute a database query',
inputSchema: z.object({
sql: z.string(),
params: z.array(z.any()).optional()
}),
outputSchema: z.object({
rows: z.array(z.record(z.any())),
rowCount: z.number()
})
},
async (args) => {
const result = await this.db.query(args.sql, args.params);
return {
content: [{
type: "text",
text: `Query returned ${result.rows.length} rows`
}],
structuredContent: {
rows: result.rows,
rowCount: result.rows.length
}
};
}
);
// Schema inspection resource
this.registerResourceHelper(
'schema.json',
'db://schema',
{
description: 'Database schema information'
},
async () => {
const schema = await this.db.getSchema();
return {
contents: [{
text: JSON.stringify(schema, null, 2),
uri: 'schema.json',
mimeType: 'application/json'
}]
};
}
);
}
}文件处理服务器
import { BaseMCPServer, BaseMCPConfig } from "./base/BaseMCPServer";
import fs from 'fs/promises';
import path from 'path';
import z from "zod";
export class FileProcessorMCP extends BaseMCPServer {
private workDir: string;
constructor(params: BaseMCPConfig & { workDir: string }) {
super(params);
this.workDir = params.workDir;
}
protected registerComponents(): void {
// Read file tool
this.registerToolHelper(
'read_file',
{
description: 'Read a file from the work directory',
inputSchema: z.object({
filename: z.string(),
encoding: z.enum(['utf8', 'base64', 'hex']).default('utf8')
}),
outputSchema: z.object({
content: z.string(),
size: z.number()
})
},
async (args) => {
const filepath = path.join(this.workDir, args.filename);
const content = await fs.readFile(filepath, args.encoding);
const stats = await fs.stat(filepath);
return {
content: [{
type: "text",
text: `File read successfully (${stats.size} bytes)`
}],
structuredContent: {
content: content.toString(),
size: stats.size
}
};
}
);
// Write file tool
this.registerToolHelper(
'write_file',
{
description: 'Write content to a file',
inputSchema: z.object({
filename: z.string(),
content: z.string(),
encoding: z.enum(['utf8', 'base64', 'hex']).default('utf8')
}),
outputSchema: z.object({
success: z.boolean(),
bytesWritten: z.number()
})
},
async (args) => {
const filepath = path.join(this.workDir, args.filename);
await fs.writeFile(filepath, args.content, args.encoding);
const stats = await fs.stat(filepath);
return {
content: [{
type: "text",
text: `File written successfully`
}],
structuredContent: {
success: true,
bytesWritten: stats.size
}
};
}
);
// Directory listing resource
this.registerResourceHelper(
'directory.json',
'file://directory/listing',
{
description: 'List files in work directory'
},
async () => {
const files = await fs.readdir(this.workDir, { withFileTypes: true });
const listing = await Promise.all(
files.map(async (file) => {
const stats = await fs.stat(path.join(this.workDir, file.name));
return {
name: file.name,
type: file.isDirectory() ? 'directory' : 'file',
size: stats.size,
modified: stats.mtime.toISOString()
};
})
);
return {
contents: [{
text: JSON.stringify(listing, null, 2),
uri: 'directory.json',
mimeType: 'application/json'
}]
};
}
);
}
}故障排除
常见问题及解决方法
1.运输未初始化
错误： “HTTP传输程序未初始化”
解决方案：
// Ensure transport mode is set correctly
const server = new YourMCP({
transportMode: "streamable-http", // Must match the transport you're trying to use
// ...
});2.工具注册失败
错误： “工具注册失败”
解决方案：
// Ensure handler is bound correctly
this.registerToolHelper(
'tool_name',
config,
this.handler.bind(this) // Don't forget .bind(this)
);3.架构验证错误
错误： “输入架构无效”
解决方案：
// Use proper Zod schemas
inputSchema: z.object({
field: z.string() // Correct
})
// Not:
inputSchema: {
field: z.string() // Missing z.object()
}4.连接问题（HTTP模式）
错误： “EADDRINUSE:地址已在使用中”
解决方案：
// Use a different port or kill the process using the port
const server = new YourMCP({
port: 3001, // Try a different port
// ...
});5.异步处理程序未等待
错误： “返回的是Promise而不是result”
解决方案：
// Mark handler as async and use await
private async myHandler(args: any): Promise {
const result = await someAsyncOperation(); // Don't forget await
return {
content: [{ type: "text", text: result }],
structuredContent: { result }
};
}调试模式
启用调试日志以进行故障排除：
export class DebugMCP extends BaseMCPServer {
private debug = process.env.DEBUG === 'true';
protected registerComponents(): void {
if (this.debug) {
console.log('[DEBUG] Registering components');
}
// ...
}
private async handler(args: any): Promise {
if (this.debug) {
console.log('[DEBUG] Handler called:', JSON.stringify(args, null, 2));
}
// ...
}
}性能监控
class PerformanceMonitor {
private metrics = new Map();
async measure(name: string, fn: () => Promise): Promise {
const start = performance.now();
try {
return await fn();
} finally {
const duration = performance.now() - start;
if (!this.metrics.has(name)) {
this.metrics.set(name, []);
}
this.metrics.get(name)!.push(duration);
if (duration > 1000) {
console.warn(`[PERF] ${name} took ${duration.toFixed(2)}ms`);
}
}
}
getStats(name: string) {
const times = this.metrics.get(name) || [];
if (times.length === 0) return null;
return {
count: times.length,
min: Math.min(...times),
max: Math.max(...times),
avg: times.reduce((a, b) => a + b, 0) / times.length
};
}
}贡献
开发设置
- 克隆存储库
- 安装依赖项：
npm install - 构建项目：
npm run build - 运行测试：
npm test - 开始开发：
npm run dev
测试您的MCP服务器
// test/YourMCP.test.ts
import { YourMCP } from '../src/servers/YourMCP';
describe('YourMCP', () => {
let server: YourMCP;
beforeEach(() => {
server = new YourMCP({
name: "TestMCP",
version: "1.0.0",
transportMode: "stdio"
});
});
test('should register tools', () => {
const metadata = server.getMetadata();
expect(metadata.name).toBe("TestMCP");
});
test('tool should return expected result', async () => {
// Test your tool handlers
});
});许可证
MIT许可证-有关详细信息，请参阅许可证文件