内容
# mcp2xiaozhi
[English](README.md) | [中文](README_CN.md)
> 将任意MCP服务器（stdio / SSE / StreamableHTTP）桥接到Xiaozhi服务器上，通过WebSocket。
[](https://pypi.org/project/mcp2xiaozhi/)
[](https://pypi.org/project/mcp2xiaozhi/)
[](LICENSE)
[](https://github.com/StanleyChanH/MCP2Xiaozhi/actions/workflows/ci.yml)
[](https://stanleychanh.github.io/MCP2Xiaozhi/)
`mcp2xiaozhi` 是一个通用的桥接工具，连接任意 [Model Context Protocol](https://modelcontextprotocol.io/) （MCP）服务器到Xiaozhi服务器。Xiaozhi服务器作为MCP客户端，通过WebSocket发送JSON-RPC工具调用，并期望收到JSON-RPC回复。该软件包接收这些帧，并将它们在协议级别上中继到您的MCP服务器，无论它运行在哪里，使用什么传输协议。
```
JSON-RPC over WebSocket (text frames)
┌──────────────────────┐ wss://… ┌──────────────────────┐
│ Xiaozhi 服务器 │ ◄──────────► │ mcp2xiaozhi 桥接 │
│ （作为MCP客户端）│ └──────────┬───────────┘
└──────────────────────┘ │ JSON-RPC
│ over MCP transport
┌──────────────────────┴───────────────┐
│ stdio │ SSE │ HTTP │
▼ ▼ ▼ ▼
本地进程 远程服务器 远程服务器
```
## 功能
- 🔄 **三种传输方式，一个桥接** — `stdio`, `sse`, 和 `streamablehttp` (`http`)，全部原生支持，无需 `mcp-proxy` 子进程。
- 🧱 **协议级中继** — 帧被解析为 `JSONRPCMessage` （封装在SDK的 `SessionMessage` 中），验证后重新序列化。格式错误的帧将被记录并丢弃，而不是导致桥接崩溃。
- 🔁 **自动重连** — 指数退避，带抖动；区分干净关闭和异常关闭。
- 🗂️ **多服务器支持** — 运行一个桥接或配置文件中的所有服务器；每个服务器都有自己的端点。
- ⚙️ **配置驱动** — 与Xiaozhi官方 `mcp-calculator` 演示的 `mcp_config.json` 格式兼容。
- 🖥️ **跨平台** — Windows上的UTF-8控制台处理，优雅的 `SIGINT` / `SIGTERM` 关闭。
- 📦 **真正的软件包** — `pyproject.toml`, 源代码布局，类型提示，CLI入口点，测试，CI，使用 [uv](https://docs.astral.sh/uv/) 管理。
## 安装
使用 [uv](https://docs.astral.sh/uv/) （推荐）：
```bash
uv tool install mcp2xiaozhi # 作为工具安装CLI
# 或，在项目中：
uv add mcp2xiaozhi
```
使用pip：
```bash
pip install mcp2xiaozhi
```
从源代码（用于开发）：
```bash
git clone https://github.com/StanleyChanH/MCP2Xiaozhi.git
cd mcp2xiaozhi
uv sync --extra dev
```
## 快速开始
1. 创建一个MCP服务器（或使用现有的服务器）。一个最小的stdio计算器：
```python
# calculator.py
from mcp.server.fastmcp import FastMCP
import math
mcp = FastMCP("Calculator")
@mcp.tool()
def calculator(python_expression: str) -> dict:
"""评估一个Python数学表达式。"""
return {"success": True, "result": eval(python_expression, {"math": math})}
if __name__ == "__main__":
mcp.run(transport="stdio")
```
2. 在 `mcp_config.json` 中描述您的服务器：
```json
{
"mcpServers": {
"calculator": {
"type": "stdio",
"command": "python",
"args": ["calculator.py"]
}
}
}
```
3. 设置Xiaozhi WebSocket端点并运行：
```bash
export MCP_ENDPOINT="wss://api.your-xiaozhi-server.example/mcp/<token>"
mcp2xiaozhi run calculator
```
或一次性运行所有启用的服务器：
```bash
mcp2xiaozhi run # 所有启用的服务器
mcp2xiaozhi list # 显示配置的服务器
mcp2xiaozhi version
```
## 配置
配置发现顺序：`--config PATH` → `$MCP_CONFIG` → `./mcp_config.json`.
### 架构
```jsonc
{
"mcpServers": {
"my-server": {
"type": "stdio", // stdio | sse | streamablehttp | http
"disabled": false, // 可选；跳过运行时运行所有服务器
// 仅限stdio
"command": "python",
"args": ["-m", "my_server"],
"env": { "FOO": "bar" }, // 与当前环境合并
// 仅限sse / streamablehttp
"url": "https://example.com/mcp",
"headers": { "Authorization": "Bearer xxx" },
"timeout": 5.0, // 连接超时（秒）
"sse_read_timeout": 300.0, // 长生命周期读取超时（秒）
// 可选的每服务器Xiaozhi端点
"endpoint": "wss://api.example.com/mcp/<token>"
}
}
}
```
### 端点解析
每个服务器都需要它应该连接到的Xiaozhi WebSocket端点。按优先级解析：
1. 服务器配置中的 `endpoint` 字段
2. `$MCP_ENDPOINT_<NAME>` 环境变量（名称大写，非字母数字字符替换为 `_`）
3. 全局 `$MCP_ENDPOINT`
当运行 **多个** 服务器时，为每个服务器提供其自己的端点 — 否则Xiaozhi服务器无法将工具调用路由到正确的服务器。如果服务器使用全局端点而其他服务器正在运行，桥接器将发出警告。
## 传输方式
| 类型 | 使用场景 | 备注 |
|------|----------|-------|
| `stdio` | 您的MCP服务器是本地脚本/二进制文件 | 作为子进程启动；环境与当前进程合并。 |
| `sse` | 使用服务器发送事件的旧版远程MCP服务器 | GET `/sse` 获取流，POST发送请求 — 由SDK处理。 |
| `streamablehttp` / `http` | 现代远程MCP服务器（推荐） | 生产HTTP传输。`http` 是别名。 |
所有三种方式都使用官方的 [`mcp` Python SDK](https://github.com/modelcontextprotocol/python-sdk) 传输原语，产生 `(read, write)` 内存流，携带 `SessionMessage` 对象。桥接器在WebSocket和这些流之间泵送消息 — 它从不启动 `mcp-proxy`。
## CLI
```
mcp2xiaozhi [--config PATH] [--log-level LEVEL] <command>
commands:
run [SERVER] 运行一个服务器（或省略/ --all运行所有启用的服务器）
list 列出配置的服务器
version 打印版本
options:
--endpoint URL 覆盖Xiaozhi端点（仅限单服务器运行）
--log-level DEBUG | INFO | WARNING | ERROR | CRITICAL
```
`python -m mcp2xiaozhi …` 也有效。
## 程序化使用
```python
import asyncio
from mcp2xiaozhi import McpBridge, load_config, resolve_endpoint, get_global_endpoint
async def main():
config = load_config()
server = config.require("calculator")
endpoint = resolve_endpoint(server, global_endpoint=get_global_endpoint())
bridge = McpBridge(server, endpoint)
await bridge.run() # 重新连接直到取消
asyncio.run(main())
```
使用 `ServerManager` 同时运行多个：
```python
from mcp2xiaozhi import ServerManager, load_config
manager = ServerManager.from_config(load_config())
asyncio.run(manager.run())
```
## 部署
桥接器是一个 **长期运行的中继** — 运行它的主机必须保持在线，
因为Xiaozhi服务器从不直接连接到您的MCP服务器。对于远程SSE/HTTP MCP服务器，您可以将其部署到任何始终在线的机器（VPS，NAS，树莓派，容器）并关闭笔记本电脑。
快速选项：
```bash
# Docker（任何平台）— 仓库提供Dockerfile + docker-compose.yml
docker compose up -d --build
# Linux systemd — 自动启动，崩溃后重启
sudo systemctl enable --now mcp2xiaozhi
# Windows — NSSM将其包装为原生服务
nssm install mcp2xiaozhi mcp2xiaozhi.exe
```
➡️ 完整指南（配置和机密，多服务器，日志，升级）： [部署文档](https://stanleychanh.github.io/MCP2Xiaozhi/deployment/).
## 与官方 `mcp-calculator` 演示的区别
| | `mcp-calculator` 演示 | `mcp2xiaozhi` |
|---|---|---|
| 传输方式 | 直接stdio；sse/http通过 `mcp-proxy` 子进程 | 所有三种原生方式，通过 `mcp` SDK |
| 分帧 | 字节管道直通 | 协议级：解析 → 验证 → 重新序列化（`SessionMessage` 感知） |
| 打包 | 单个脚本 | PyPI软件包，CLI，源代码布局 |
| 重连 | 每服务器退避 | 每桥接退避，带抖动 + 清洁/异常关闭区分 |
| 多服务器 | 所有服务器共享一个端点 | 每服务器端点，冲突警告 |
| 错误处理 | 引发 + 重连 | 格式错误的帧被丢弃；结构化的异常组解包 |
## 开发
```bash
uv sync --extra dev # 安装开发依赖
uv run ruff check . # 校验
uv run mypy src # 类型检查
uv run pytest # 测试
uv build # 构建sdist + wheel
```
参见 [CONTRIBUTING.md](CONTRIBUTING.md).
## 许可证
MIT — 参见 [LICENSE](LICENSE).
