mcp2xiaozhi 0.1.0

Bridge any MCP server (stdio / SSE / StreamableHTTP) to a Xiaozhi server over WebSocket.

mcp2xiaozhi

English | 中文

Bridge any MCP server (stdio / SSE / StreamableHTTP) to a Xiaozhi server over WebSocket.

mcp2xiaozhi is a general-purpose bridge that connects any Model Context Protocol (MCP) server to a Xiaozhi server. The Xiaozhi server acts as an MCP client over a WebSocket: it sends JSON-RPC tool calls as text frames and expects JSON-RPC replies. This package receives those frames and relays them — at the protocol level — to your MCP server, wherever it runs and whatever transport it speaks.

                 JSON-RPC over WebSocket (text frames)
   ┌──────────────────────┐   wss://…    ┌──────────────────────┐
   │   Xiaozhi server     │ ◄──────────► │   mcp2xiaozhi bridge │
   │  (acts as MCP client)│              └──────────┬───────────┘
   └──────────────────────┘                          │  JSON-RPC
                                                     │  over MCP transport
                              ┌──────────────────────┴───────────────┐
                              │   stdio        │   SSE   │   HTTP    │
                              ▼                ▼         ▼           ▼
                         local process    remote server          remote server

Features

• 🔄 Three transports, one bridge — stdio, sse, and streamablehttp (http), all native, no mcp-proxy subprocess required.
• 🧱 Protocol-level relay — frames are parsed as JSONRPCMessage (wrapped in the SDK's SessionMessage), validated, then re-serialized. Malformed frames are logged and dropped instead of crashing the bridge.
• 🔁 Automatic reconnection — exponential backoff with jitter on either side dropping; clean closes distinguished from abnormal ones.
• 🗂️ Multi-server — run one bridge or every server in your config; each server gets its own endpoint.
• ⚙️ Config-driven — drop-in compatible with the mcp_config.json shape from Xiaozhi's official mcp-calculator demo.
• 🖥️ Cross-platform — UTF-8 console handling on Windows, graceful SIGINT/SIGTERM shutdown.
• 📦 Real package — pyproject.toml, src layout, type hints, CLI entry point, tests, CI, managed with uv.

Install

With uv (recommended):

uv tool install mcp2xiaozhi        # install the CLI as a tool
# or, in a project:
uv add mcp2xiaozhi

With pip:

pip install mcp2xiaozhi

From source (for development):

git clone https://github.com/StanleyChanH/MCP2Xiaozhi.git
cd mcp2xiaozhi
uv sync --extra dev

Quick start

1. Create an MCP server (or use an existing one). A minimal stdio calculator:

   # calculator.py
   from mcp.server.fastmcp import FastMCP
   import math
   
   mcp = FastMCP("Calculator")
   
   @mcp.tool()
   def calculator(python_expression: str) -> dict:
       """Evaluate a Python math expression."""
       return {"success": True, "result": eval(python_expression, {"math": math})}
   
   if __name__ == "__main__":
       mcp.run(transport="stdio")
   

2. Describe your servers in mcp_config.json:

   {
     "mcpServers": {
       "calculator": {
         "type": "stdio",
         "command": "python",
         "args": ["calculator.py"]
       }
     }
   }
   

3. Set the Xiaozhi WebSocket endpoint and run:

   export MCP_ENDPOINT="wss://api.your-xiaozhi-server.example/mcp/<token>"
   mcp2xiaozhi run calculator
   

   Or run every enabled server at once:

   mcp2xiaozhi run            # all enabled servers
   mcp2xiaozhi list           # show configured servers
   mcp2xiaozhi version
   

Configuration

Config discovery order: --config PATH → $MCP_CONFIG → ./mcp_config.json.

Schema

{
  "mcpServers": {
    "my-server": {
      "type": "stdio",              // stdio | sse | streamablehttp | http
      "disabled": false,            // optional; skip when running all

      // stdio-only
      "command": "python",
      "args": ["-m", "my_server"],
      "env": { "FOO": "bar" },      // merged onto the current environment

      // sse / streamablehttp-only
      "url": "https://example.com/mcp",
      "headers": { "Authorization": "Bearer xxx" },
      "timeout": 5.0,               // connect timeout (s)
      "sse_read_timeout": 300.0,    // long-lived read timeout (s)

      // optional per-server Xiaozhi endpoint
      "endpoint": "wss://api.example.com/mcp/<token>"
    }
  }
}

Endpoint resolution

Each server needs the Xiaozhi WebSocket endpoint it should connect to. Resolved in priority:

1. endpoint field in the server config
2. $MCP_ENDPOINT_<NAME> environment variable (name uppercased, non-alphanumeric → _)
3. global $MCP_ENDPOINT

When running multiple servers, give each its own endpoint — otherwise the Xiaozhi server cannot route tool calls to the right server. The bridge will warn if a server falls back to the global endpoint while others are running.

Transports

Type	Use when	Notes
stdio	Your MCP server is a local script/binary	Spawns it as a child process; env merged with the current process.
sse	Legacy remote MCP server using Server-Sent Events	GET /sse for the stream, POST for requests — handled by the SDK.
streamablehttp / http	Modern remote MCP server (recommended)	The production HTTP transport. http is an alias.

All three are implemented with the official mcp Python SDK transport primitives, which yield (read, write) memory streams carrying SessionMessage objects. The bridge pumps messages between the WebSocket and those streams — it never spawns mcp-proxy.

CLI

mcp2xiaozhi [--config PATH] [--log-level LEVEL] <command>

commands:
  run [SERVER]       Run one server (or all enabled if omitted / --all)
  list               List configured servers
  version            Print version

options:
  --endpoint URL     Override the Xiaozhi endpoint (single-server runs only)
  --log-level        DEBUG | INFO | WARNING | ERROR | CRITICAL

python -m mcp2xiaozhi … works too.

Programmatic usage

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()   # reconnects forever until cancelled

asyncio.run(main())

Run several at once with ServerManager:

from mcp2xiaozhi import ServerManager, load_config

manager = ServerManager.from_config(load_config())
asyncio.run(manager.run())

How it differs from the official mcp-calculator demo

	mcp-calculator demo	mcp2xiaozhi
Transports	stdio directly; sse/http via mcp-proxy subprocess	All three native via the mcp SDK
Framing	Byte-pipe passthrough	Protocol-level: parse → validate → re-serialize (SessionMessage aware)
Packaging	Single script	PyPI package, CLI, src layout
Reconnect	Per-server backoff	Per-bridge backoff with jitter + clean/abnormal close distinction
Multi-server	All share one endpoint	Per-server endpoints with conflict warnings
Error handling	Raises + reconnects	Malformed frames dropped; structured exception-group unwrapping

Development

uv sync --extra dev          # install dev dependencies
uv run ruff check .          # lint
uv run mypy src              # type check
uv run pytest                # tests
uv build                     # build sdist + wheel

See CONTRIBUTING.md.

License

MIT — see LICENSE.