MCP服务器

Name: MCP服务器
Author: freedanfan

作者：freedanfan10 星标更新：2025-05-02

安装使用来源：spider==github.com

项目介绍

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             # 项目文档

安装

克隆仓库：

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

安装依赖：

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协议实现

初始化流程

客户端通过SSE连接到服务器
服务器返回API端点URI
客户端发送带有协议版本和能力的初始化请求
服务器响应初始化请求，返回服务器能力

采样请求

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

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "method": "sample",
  "params": {
    "prompt": "你好，请介绍一下自己。"
  }
}

服务器将返回采样结果：

{
  "jsonrpc": "2.0",
  "id": "request-id",
  "result": {
    "content": "这是对提示的响应...",
    "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:
    """处理新方法"""
    logger.info(f"收到新方法请求：{params}")
    # 处理逻辑
    return {"result": "成功"}

def _register_methods(self):
    # 注册现有方法
    self.router.register_method("initialize", self.handle_initialize)
    self.router.register_method("sample", self.handle_sample)
    self.router.register_method("shutdown", self.handle_shutdown)
    # 注册新方法
    self.router.register_method("new_method", self.handle_new_method)

集成AI模型

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

async def handle_sample(self, params: dict) -> dict:
    """处理采样请求"""
    logger.info(f"收到采样请求：{params}")
    
    # 获取提示
    prompt = params.get("prompt", "")
    
    # 调用AI模型API
    # 例如：使用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
        }
    }

故障排除

常见问题

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

日志记录

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

# 提高日志级别
export PYTHONPATH=.
python -m logging -v DEBUG -m mcp_server

参考资料

许可证

本项目采用MIT许可证。详情参见LICENSE文件。