模块化MCP服务器
一个可扩展的、支持插件的MCP(模型上下文协议)服务器,具有人在环协作功能。
特性
- 模块化架构:插件就绪操作注册表,便于扩展
- 循环中的人类:敏感操作的内置审批工作流
- 异步支持:aiohttp完全支持异步/等待
- 综合录井:具有唯一请求跟踪的详细日志记录
- 调试支持:每个操作都有唯一的ID用于可追溯性
- 生产就绪:具有错误处理和监控功能的可扩展设计
建筑
mcp_server/
├── core/ # Core server components
│ ├── server.py # Main MCPServer class
│ ├── action_registry.py # Action management system
│ └── human_loop.py # Human approval workflow
├── actions/ # Action implementations
│ ├── client_data.py # Client data actions
│ └── ... # Additional action modules
├── utils/ # Utility functions
├── config/ # Configuration management
└── __init__.py # Package initialization快速开始
- 再进行:
pip install -r requirements.txt- 启动服务器:
python main.py- 测试服务器:
# Health check
curl http://localhost:8000/health
# Get client data
curl -X POST http://localhost:8000/api/action/get_client_data \
-H "Content-Type: application/json" \
-d '{"params": {"client_id": "test-client"}}'API终点
核心终点
GET /health-健康检查GET /api/status-服务器状态和调试信息GET /api/actions-列出可用操作
动作执行
POST /api/action/{action_name}-执行操作POST /api/approval/{request_id}-处理人工审批
人在循环工作流程
可以将操作配置为需要人工批准:
- 行动注册:设置
requires_approval=True注册操作时 - 申请批复:服务器在调用操作时创建审批请求
- 人工审核:人工审核并批准/拒绝该操作
- 执行:已批准的操作将自动执行
发展
添加新操作
- 在中创建操作
mcp_server/actions/ - 使用
@action_handler元数据装饰器 - 在中注册操作
main.py
示例操作
from mcp_server.core.action_registry import action_handler
@action_handler(
description="My custom action",
requires_approval=True
)
async def my_action(params: Dict[str, Any]) -> Dict[str, Any]:
# Action implementation
return {"result": "success"}配置
服务器支持通过环境变量进行配置:
MCP_HOST:服务器主机(默认:localhost)MCP_PORT:服务器端口(默认值:8000)MCP_DEBUG:启用调试模式(默认值:false)
日志记录
服务器提供全面的日志记录:
- 请求跟踪:每个请求都有一个唯一的ID
- 性能监控:执行时间跟踪
- 错误处理:带上下文的详细错误日志记录
- 人体循环:审批工作流日志记录
生产部署
对于生产部署:
- 使用流程管理器(systemd、supervisor)
- 配置反向代理(nginx、Apache)
- 设置监控和警报
- 配置SSL/TLS证书
- 设置备份和恢复程序
贡献
- 遵循模块化架构
- 添加全面的文档
- 包括所有操作的唯一ID
- 使用异步操作进行彻底测试
- 对于敏感操作,遵循人在循环工作流程
