MCP 服务器
中文詳細
项目概述
该项目基于 FastAPI 和 MCP（模型上下文协议）构建，实现了 AI 模型与开发环境之间标准化的上下文交互。通过简化模型部署、提供高效的 API 端点以及确保模型输入输出的一致性，增强了 AI 应用的可扩展性和可维护性，使开发者能够更轻松地集成和管理 AI 任务。
MCP（模型上下文协议）是 AI 模型与开发环境之间上下文交互的统一协议。本项目提供基于 Python 的 MCP 服务器实现，支持 MCP 协议的基本功能，包括初始化、采样和会话管理。
特征
- JSON-RPC 2.0 ：基于标准 JSON-RPC 2.0 协议的请求-响应通信
- SSE 连接：支持服务器发送事件连接以实现实时通知
- 模块化设计：模块化架构，易于扩展和定制
- 异步处理：使用 FastAPI 和异步 IO 的高性能服务
- 完整客户端：包括完整的测试客户端实现
项目结构
mcp_server/
├── mcp_server.py # MCP server main program
├── mcp_client.py # MCP client test program
├── routers/
│ ├── __init__.py # Router package initialization
│ └── base_router.py # Base router implementation
├── requirements.txt # Project dependencies
└── README.md # Project documentation
安装
- 克隆存储库：
git clone https://github.com/freedanfan/mcp_server.git
cd mcp_server
- 安装依赖项：
pip install -r requirements.txt
用法
启动服务器
默认情况下，服务器将从127.0.0.1:12000启动。您可以使用环境变量自定义主机和端口：
export MCP_SERVER_HOST=0.0.0.0
export MCP_SERVER_PORT=8000
python mcp_server.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": "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 模型，请修改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
}
}
故障排除
常见问题
- 连接错误：确保服务器正在运行，并且客户端正在使用正确的服务器 URL
- 405 方法不允许：确保客户端将请求发送到正确的 API 端点
- SSE 连接失败：检查网络连接和防火墙设置
日志记录
服务器和客户端均提供详细的日志记录。查看日志以获取更多信息：
# Increase log level
export PYTHONPATH=.
python -m logging -v DEBUG -m mcp_server
参考
执照
本项目遵循 MIT 许可证。详情请参阅 LICENSE 文件。