mcp_router/
├── main.py                 # 项目入口
├── config.json            # 全局配置文件
├── requirements.txt       # 依赖文件
├── pyproject.toml         # uv项目配置
│
├── src/
│   ├── core/              # 核心模块
│   │   ├── logger.py      # 日志系统
│   │   ├── config.py      # 配置管理器
│   │   └── exceptions.py  # 自定义异常
│   │
│   ├── mcp/               # MCP模块
│   │   ├── client.py      # MCP客户端管理
│   │   ├── server.py      # MCP服务端
│   │   ├── router.py      # 路由核心逻辑
│   │   └── transport.py   # 传输层
│   │
│   ├── api/               # API模块
│   │   ├── app.py         # FastAPI应用
│   │   └── routes.py      # API路由处理
│   │
│   └── utils/             # 工具模块
│       ├── validator.py   # 输入验证
│       ├── watcher.py     # 文件监视器
│       └── security.py    # 安全工具
│
├── data/                  # MCP配置目录
│   ├── example/
│   │   └── mcp_settings.json
│   └── ...
│
└── test/                  # 测试文件
    ├── test_router.py
    ├── test_api.py
    └── test_security.py

# 使用 uv (推荐)
uv venv .venv
uv pip install -e ".[dev]"

# 或使用 pip
python -m venv .venv
.venv\Scripts\activate  # Windows
source .venv/bin/activate  # Linux/Mac
pip install -e ".[dev]"

# 或使用 conda
conda env create -f environment.yml
conda activate mcp_router

• server.host: HTTP/SSE模式的监听地址（默认：0.0.0.0）
• server.http.port: HTTP模式的监听端口（默认：3000）
• server.sse.port: SSE模式的监听端口（默认：3001）
• server.allow_instance_management: 允许LLM管理实例（默认：false）
• api.enabled: 是否启动REST API服务器（默认：false）
• api.port: REST API端口（默认：8001）
• api.auto_find_port: 端口占用时自动递增查找可用端口
• api.enable_realtime_logs: 启用WebSocket实时日志 (ws://host:port/ws)
• logging.directory: 日志目录，使用Minecraft风格 (latest-{mode}.txt + 时间戳备份)
• logging.level: 日志级别（DEBUG/INFO/WARNING/ERROR/OFF）

注意：传输模式（stdio/http/sse）通过命令行参数指定，不在配置文件中设置。

当 API 模式启用时，可通过以下端点管理 MCP Router：

实例管理:

• GET /api/instances - 列出所有实例
• GET /api/instances/{name} - 获取实例详情
• POST /api/instances - 添加新实例
• PATCH /api/instances/{name} - 更新实例配置
• DELETE /api/instances/{name} - 删除实例
• POST /api/instances/{name}/enable - 启用实例
• POST /api/instances/{name}/disable - 禁用实例

工具管理:

• GET /api/tools - 列出所有工具
• GET /api/tools/{instance_name} - 获取实例的工具列表
• POST /api/call - 调用工具

其他:

• GET / - 服务状态
• GET /health - 健康检查
• GET /api/config - 获取配置
• WS /ws - 实时日志流 (需启用 enable_realtime_logs)

curl 示例:

# 初始化会话
curl -X POST http://localhost:3000/mcp \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":0,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}'

# 列出工具
curl -X POST http://localhost:3000/mcp \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'

Python 客户端示例:

import httpx

class MCPHTTPClient:
    def __init__(self, base_url="http://localhost:3000"):
        self.client = httpx.Client()
        self.base_url = base_url
        self.request_id = 0
        self.initialized = False
    
    def call_method(self, method, params=None):
        self.request_id += 1
        response = self.client.post(f"{self.base_url}/mcp", json={
            "jsonrpc": "2.0",
            "id": self.request_id,
            "method": method,
            "params": params or {}
        })
        return response.json()
    
    def initialize(self):
        result = self.call_method("initialize", {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "python-client", "version": "1.0.0"}
        })
        self.initialized = True
        return result
    
    def list_tools(self):
        if not self.initialized:
            self.initialize()
        return self.call_method("tools/list")

# 使用示例
client = MCPHTTPClient()
tools = client.list_tools()
print(tools)

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";

const transport = new SSEClientTransport(
  new URL("http://localhost:3001/sse")
);
const client = new Client(
  { name: "my-client", version: "1.0.0" }, 
  { capabilities: {} }
);

await client.connect(transport);
const tools = await client.listTools();
console.log(tools);

curl 示例:

# 列出所有实例
curl http://localhost:8001/api/instances \
  -H "Authorization: Bearer your-token"

# 调用工具
curl -X POST http://localhost:8001/api/call \
  -H "Authorization: Bearer your-token" \
  -H "Content-Type: application/json" \
  -d '{"instance": "openai_doc", "tool": "read_project_oas", "params": {}}'

Python 客户端示例:

import httpx

client = httpx.Client(
    base_url="http://localhost:8001",
    headers={"Authorization": "Bearer your-token"}
)

# 列出实例
instances = client.get("/api/instances").json()

# 调用工具
result = client.post("/api/call", json={
    "instance": "openai_doc",
    "tool": "read_project_oas",
    "params": {}
}).json()

# 运行所有测试
pytest

# 运行特定测试
pytest test/test_router.py

# 带覆盖率
pytest --cov=src --cov-report=html