MCP 服务器

中文文档

项目概览

本项目基于 FastAPI 和 MCP (模型上下文协议) 构建，实现了 AI 模型与开发环境之间标准化的上下文交互。它通过简化模型部署、提供高效的 API 接口以及确保模型输入输出的一致性，增强了 AI 应用的可扩展性和可维护性，从而使开发者能够更轻松地集成和管理 AI 任务。

MCP (模型上下文协议) 是 AI 模型与开发环境之间上下文交互的统一协议。本项目提供了一个基于 Python 的 MCP 服务器实现，支持基本的 MCP 协议功能，包括初始化、采样和会话管理。

特性

• JSON-RPC 2.0: 基于标准 JSON-RPC 2.0 协议的请求-响应通信
• SSE 连接: 支持服务器发送事件 (Server-Sent Events) 连接，用于实时通知
• 模块化设计: 模块化架构，易于扩展和定制
• 异步处理: 使用 FastAPI 和异步 IO 实现高性能服务
• 完整客户端: 包含完整的测试客户端实现

项目结构

mcp_server/
├── mcp_server.py         # MCP 服务器主程序
├── mcp_client.py         # MCP 客户端测试程序
├── routers/
│   ├── __init__.py       # 路由包初始化
│   └── base_router.py    # 基础路由实现
├── requirements.txt      # 项目依赖
└── README.md             # 项目文档

安装

1. 克隆仓库:

git clone https://github.com/freedanfan/mcp_server.git
cd mcp_server

2. 安装依赖:

pip install -r requirements.txt

使用

启动服务器

python mcp_server.py

默认情况下，服务器将在 127.0.0.1:12000 上启动。您可以使用环境变量自定义主机和端口：

export MCP_SERVER_HOST=0.0.0.0
export MCP_SERVER_PORT=8000
python mcp_server.py

运行客户端

在另一个终端中运行客户端：

python mcp_client.py

如果服务器未在默认地址运行，您可以设置一个环境变量：

export MCP_SERVER_URL="http://your-server-address:port"
python mcp_client.py

API 接口

服务器提供以下 API 接口：

• 根路径 (/): 提供服务器信息
• API 接口 (/api): 处理 JSON-RPC 请求
• SSE 接口 (/sse): 处理 SSE 连接

MCP 协议实现

初始化流程

1. 客户端通过 SSE 连接到服务器
2. 服务器返回 API 接口 URI
3. 客户端发送包含协议版本和功能的初始化请求
4. 服务器响应初始化请求，返回服务器功能

采样请求

客户端可以发送带有提示的采样请求：

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "method": "sample",
  "params": {
    "prompt": "Hello, please introduce yourself."
  }
}

服务器将返回采样结果：

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "result": {
    "content": "This is a response to the prompt...",
    "usage": {
      "prompt_tokens": 10,
      "completion_tokens": 50,
      "total_tokens": 60
    }
  }
}

关闭会话

客户端可以发送关闭请求：

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "method": "shutdown",
  "params": {}
}

服务器将优雅地关闭：

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "result": {
    "status": "shutting_down"
  }
}

开发扩展

添加新方法

要添加新的 MCP 方法，请将处理函数添加到 MCPServer 类，并在 _register_methods 方法中注册它：

def handle_new_method(self, params: dict) -> dict:
    """Handle new method"""
    logger.info(f"Received new method request: {params}")
    # Processing logic
    return {"result": "success"}

def _register_methods(self):
    # Register existing methods
    self.router.register_method("initialize", self.handle_initialize)
    self.router.register_method("sample", self.handle_sample)
    self.router.register_method("shutdown", self.handle_shutdown)
    # Register new method
    self.router.register_method("new_method", self.handle_new_method)

集成 AI 模型

要集成实际的 AI 模型，请修改 handle_sample 方法：

async def handle_sample(self, params: dict) -> dict:
    """Handle sampling request"""
    logger.info(f"Received sampling request: {params}")
    
    # Get prompt
    prompt = params.get("prompt", "")
    
    # Call AI model API
    # For example: using OpenAI API
    response = await openai.ChatCompletion.acreate(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    
    content = response.choices[0].message.content
    usage = response.usage
    
    return {
        "content": content,
        "usage": {
            "prompt_tokens": usage.prompt_tokens,
            "completion_tokens": usage.completion_tokens,
            "total_tokens": usage.total_tokens
        }
    }

故障排除

常见问题

1. 连接错误: 确保服务器正在运行，并且客户端正在使用正确的服务器 URL
2. 405 方法不允许: 确保客户端正在向正确的 API 接口发送请求
3. SSE 连接失败: 检查网络连接和防火墙设置

日志

服务器和客户端都提供详细的日志记录。查看日志以获取更多信息：

# Increase log level
export PYTHONPATH=.
python -m logging -v DEBUG -m mcp_server

参考

• MCP 协议规范
• FastAPI 文档
• JSON-RPC 2.0 规范
• SSE 规范

许可证

本项目采用 MIT 许可证。有关详细信息，请参见 LICENSE 文件。