mcp协议官方文档

主题

MCP 服务器规范 ​

本节包含模型上下文协议 (MCP) 服务器实现的详细规范。

概述 ​

MCP 服务器是提供工具、资源和提示的应用程序。服务器负责：

  • 接受客户端连接
  • 处理客户端请求
  • 提供工具、资源和提示
  • 管理会话状态

服务器类型 ​

1. 工具服务器 ​

提供可执行工具的服务器：

  • 文件系统工具: 文件读写、目录操作
  • 数据库工具: 查询、更新数据库
  • API 工具: 调用外部 API
  • 计算工具: 数学计算、数据处理

2. 资源服务器 ​

提供数据资源的服务器：

  • 文档服务器: 提供文档内容
  • 数据服务器: 提供结构化数据
  • 媒体服务器: 提供图片、音频、视频
  • 配置服务器: 提供配置信息

3. 提示服务器 ​

提供提示模板的服务器：

  • 模板服务器: 提供可重用的提示模板
  • 工作流服务器: 提供复杂的工作流程
  • 助手服务器: 提供专门的助手配置

核心组件 ​

服务器接口 ​

typescript
interface MCPServer {
    // 启动服务器
    start(): Promise<void>;
    
    // 停止服务器
    stop(): Promise<void>;
    
    // 处理客户端连接
    handleConnection(transport: Transport): Promise<void>;
    
    // 获取服务器信息
    getServerInfo(): ServerInfo;
}

请求处理器 ​

typescript
interface RequestHandler {
    // 处理初始化请求
    handleInitialize(params: InitializeParams): Promise<InitializeResult>;
    
    // 处理工具相关请求
    handleListTools(): Promise<ListToolsResult>;
    handleCallTool(params: CallToolParams): Promise<CallToolResult>;
    
    // 处理资源相关请求
    handleListResources(): Promise<ListResourcesResult>;
    handleReadResource(params: ReadResourceParams): Promise<ReadResourceResult>;
    
    // 处理提示相关请求
    handleListPrompts(): Promise<ListPromptsResult>;
    handleGetPrompt(params: GetPromptParams): Promise<GetPromptResult>;
}

能力管理 ​

typescript
interface CapabilityManager {
    // 声明服务器能力
    declareCapabilities(): ServerCapabilities;
    
    // 检查能力支持
    supportsCapability(capability: string): boolean;
    
    // 更新能力
    updateCapabilities(capabilities: ServerCapabilities): void;
}

实现要求 ​

必需功能 ​

  • [x] JSON-RPC 2.0 协议支持
  • [x] 初始化握手处理
  • [x] 错误响应
  • [x] 基本能力声明

推荐功能 ​

  • [x] 工具提供
  • [x] 资源提供
  • [x] 提示提供
  • [x] 进度报告

可选功能 ​

  • [ ] 流式响应
  • [ ] 批量处理
  • [ ] 缓存支持
  • [ ] 监控指标

规范文档 ​

基础规范 ​

  • 服务器架构 - 服务器架构设计
  • 初始化流程 - 服务器初始化
  • 请求处理 - 请求处理机制

功能规范 ​

  • 工具提供 - 工具实现规范
  • 资源提供 - 资源实现规范
  • 提示提供 - 提示实现规范

高级功能 ​

  • 流式处理 - 流式响应实现
  • 批量处理 - 批量请求处理
  • 缓存机制 - 缓存实现

运维规范 ​

  • 监控指标 - 监控和指标
  • 日志记录 - 日志记录规范
  • 性能优化 - 性能优化指南

实现示例 ​

Python 服务器 ​

python
import asyncio
from mcp import Server, StdioTransport
from mcp.types import Tool, TextContent

# 创建服务器实例
server = Server("my-server")

# 定义工具
@server.tool()
async def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        # 安全的表达式计算
        result = eval(expression, {"__builtins__": {}})
        return f"结果: {result}"
    except Exception as e:
        return f"错误: {str(e)}"

@server.tool()
async def get_weather(city: str) -> str:
    """获取天气信息"""
    # 模拟天气 API 调用
    return f"{city} 的天气: 晴朗，温度 25°C"

# 定义资源
@server.resource("config://settings")
async def get_settings() -> str:
    """获取配置设置"""
    return "server_name: my-server\nversion: 1.0.0\ndebug: false"

# 定义提示
@server.prompt()
async def code_review_prompt(language: str, code: str) -> str:
    """代码审查提示"""
    return f"""
请审查以下 {language} 代码：

```{language}
{code}

请检查：

  1. 代码质量
  2. 潜在问题
  3. 改进建议 """

async def main(): # 创建传输层 transport = StdioTransport()

# 运行服务器
await server.run(transport)

if name == "main": asyncio.run(main())


### TypeScript 服务器
```typescript
import { Server, StdioTransport } from "@modelcontextprotocol/sdk";
import { Tool, Resource, Prompt } from "@modelcontextprotocol/sdk/types.js";

// 创建服务器
const server = new Server({
    name: "my-server",
    version: "1.0.0"
});

// 工具处理器
server.setRequestHandler("tools/list", async () => {
    return {
        tools: [
            {
                name: "file_read",
                description: "读取文件内容",
                inputSchema: {
                    type: "object",
                    properties: {
                        path: { type: "string", description: "文件路径" }
                    },
                    required: ["path"]
                }
            },
            {
                name: "file_write",
                description: "写入文件内容",
                inputSchema: {
                    type: "object",
                    properties: {
                        path: { type: "string", description: "文件路径" },
                        content: { type: "string", description: "文件内容" }
                    },
                    required: ["path", "content"]
                }
            }
        ]
    };
});

server.setRequestHandler("tools/call", async (request) => {
    const { name, arguments: args } = request.params;
    
    switch (name) {
        case "file_read":
            try {
                const fs = await import("fs/promises");
                const content = await fs.readFile(args.path, "utf-8");
                return {
                    content: [{ type: "text", text: content }]
                };
            } catch (error) {
                return {
                    isError: true,
                    content: [{ 
                        type: "text", 
                        text: `读取文件失败: ${error.message}` 
                    }]
                };
            }
            
        case "file_write":
            try {
                const fs = await import("fs/promises");
                await fs.writeFile(args.path, args.content, "utf-8");
                return {
                    content: [{ 
                        type: "text", 
                        text: `文件已写入: ${args.path}` 
                    }]
                };
            } catch (error) {
                return {
                    isError: true,
                    content: [{ 
                        type: "text", 
                        text: `写入文件失败: ${error.message}` 
                    }]
                };
            }
            
        default:
            return {
                isError: true,
                content: [{ 
                    type: "text", 
                    text: `未知工具: ${name}` 
                }]
            };
    }
});

// 资源处理器
server.setRequestHandler("resources/list", async () => {
    return {
        resources: [
            {
                uri: "file://README.md",
                name: "项目说明",
                description: "项目的 README 文件",
                mimeType: "text/markdown"
            }
        ]
    };
});

server.setRequestHandler("resources/read", async (request) => {
    const { uri } = request.params;
    
    if (uri === "file://README.md") {
        try {
            const fs = await import("fs/promises");
            const content = await fs.readFile("README.md", "utf-8");
            return {
                contents: [{ 
                    uri, 
                    mimeType: "text/markdown", 
                    text: content 
                }]
            };
        } catch (error) {
            throw new Error(`读取资源失败: ${error.message}`);
        }
    }
    
    throw new Error(`未知资源: ${uri}`);
});

// 启动服务器
async function main() {
    const transport = new StdioTransport();
    await server.connect(transport);
    console.error("MCP 服务器已启动");
}

main().catch(console.error);

测试指南 ​

单元测试 ​

python
import pytest
from unittest.mock import Mock, AsyncMock
from mcp import Server

@pytest.mark.asyncio
async def test_tool_call():
    server = Server("test-server")
    
    @server.tool()
    async def test_tool(param: str) -> str:
        return f"处理: {param}"
    
    # 模拟工具调用
    result = await server.call_tool("test_tool", {"param": "测试"})
    assert result.content[0].text == "处理: 测试"

@pytest.mark.asyncio
async def test_resource_read():
    server = Server("test-server")
    
    @server.resource("test://resource")
    async def test_resource() -> str:
        return "测试资源内容"
    
    # 模拟资源读取
    result = await server.read_resource("test://resource")
    assert result.contents[0].text == "测试资源内容"

集成测试 ​

python
import asyncio
import subprocess
from mcp import Client, StdioTransport

async def test_server_integration():
    # 启动服务器进程
    server_process = subprocess.Popen([
        "python", "-m", "my_server"
    ])
    
    try:
        # 连接到服务器
        transport = StdioTransport("python", ["-m", "my_server"])
        client = Client("test-client")
        
        await client.connect(transport)
        await client.initialize()
        
        # 测试工具列表
        tools = await client.list_tools()
        assert len(tools.tools) > 0
        
        # 测试工具调用
        result = await client.call_tool(
            tools.tools[0].name,
            {"param": "test"}
        )
        assert result.content
        
        await client.disconnect()
        
    finally:
        server_process.terminate()
        server_process.wait()

最佳实践 ​

架构设计 ​

  1. 模块化: 将功能分解为独立模块
  2. 可扩展: 支持插件和扩展
  3. 可配置: 提供灵活的配置选项
  4. 可测试: 编写全面的测试

性能优化 ​

  1. 异步处理: 使用异步 I/O
  2. 连接池: 复用数据库连接
  3. 缓存策略: 缓存频繁访问的数据
  4. 资源管理: 及时释放资源

安全考虑 ​

  1. 输入验证: 验证所有输入参数
  2. 权限控制: 实施细粒度权限
  3. 错误处理: 安全的错误响应
  4. 日志记录: 记录安全相关事件

监控运维 ​

  1. 健康检查: 实现健康检查端点
  2. 指标收集: 收集性能指标
  3. 日志管理: 结构化日志记录
  4. 告警机制: 设置关键指标告警

部署指南 ​

Docker 部署 ​

dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

EXPOSE 8080

CMD ["python", "-m", "my_server"]

Kubernetes 部署 ​

yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mcp-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: mcp-server
  template:
    metadata:
      labels:
        app: mcp-server
    spec:
      containers:
      - name: mcp-server
        image: my-mcp-server:latest
        ports:
        - containerPort: 8080
        env:
        - name: MCP_LOG_LEVEL
          value: "INFO"
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "256Mi"
            cpu: "200m"

相关资源 ​

🚀 探索模型上下文协议的无限可能 | 连接 AI 与世界的桥梁 | 让智能更智能

Copyright © 2025 mcp协议官方文档 ｜ sitemap