da2-xiaozhi-tunnel 0.1.4

AI xiaozhi 小智 WebSocket MCP (Model Context Protocol) 橋接模組，支援 WebSocket 與 HTTP/SSE 傳輸

da2-xiaozhi-tunnel

MCP (Model Context Protocol) 橋接模組，用於連接小智雲端與其他 MCP 伺服器。

作者: Danny | 工作室: DA2 Studio | 網站: https://da2.35g.tw

---

功能特色

• WebSocketTunnelMCP: WebSocket 隧道伺服器，對接小智雲端
• MCPHttpClient: HTTP/SSE 客戶端，連接遠端 MCP 伺服器
• 支援多伺服器工具聚合與智慧路由
• 自動重連機制

---

安裝

pip install da2-xiaozhi-tunnel

---

快速開始

基礎範例 (僅使用 WebSocketTunnelMCP)

連接小智雲端，註冊本地工具供 AI 呼叫：

import asyncio
from da2_xiaozhi_tunnel import WebSocketTunnelMCP

TOKEN_URL = "wss://api.xiaozhi.me/mcp/?token=YOUR_TOKEN_HERE"

mcp = WebSocketTunnelMCP(server_name="MyServer")

# 連線事件
@mcp.on_connected()
async def on_connected():
    print("✅ 已連線")

@mcp.on_disconnected()
async def on_disconnected():
    print("❌ 連線中斷")

# 註冊工具
@mcp.tool(
    name="get_greeting",
    description="取得問候語",
    input_schema={
        "type": "object",
        "properties": {
            "name": {"type": "string", "description": "名稱"}
        },
        "required": ["name"]
    }
)
async def get_greeting(name: str):
    return f"哈囉 {name}！"

# 主程式
async def main():
    while True:
        if await mcp.connect(TOKEN_URL):
            await mcp.listen_loop()
        await asyncio.sleep(5)  # 重連間隔

asyncio.run(main())

---

橋接範例 (WebSocketTunnelMCP + MCPHttpClient)

連接小智雲端，同時橋接遠端 MCP Server，聚合所有工具：

import asyncio
from da2_xiaozhi_tunnel import WebSocketTunnelMCP, MCPHttpClient

TOKEN_URL = "wss://api.xiaozhi.me/mcp/?token=YOUR_TOKEN_HERE"
EXTERNAL_MCP_URL = "http://your-mcp-server/mcp"

mcp = WebSocketTunnelMCP(server_name="BridgeServer")

# 建立外部 MCP Client
external_client = MCPHttpClient(
    base_url=EXTERNAL_MCP_URL,
    client_name="hub",  # 工具將有 hub_ 前綴
    use_sse=False       # 純 HTTP 模式
)

# 連線事件
@mcp.on_connected()
async def on_connected():
    print(f"✅ 已連線 | 本地工具: {len(mcp.tools)} | 外部工具: {len(mcp.external_tools)}")

# 本地工具
@mcp.tool("local_echo", "本地回聲", {"type": "object", "properties": {"msg": {"type": "string"}}})
async def local_echo(msg: str):
    return f"[回聲] {msg}"

# 主程式
async def main():
    # 掛載外部 Client
    mcp.mount_client(external_client)
    
    while True:
        if await mcp.connect(TOKEN_URL):
            await mcp.listen_loop()
        await asyncio.sleep(5)

asyncio.run(main())

橋接後，AI 可以同時呼叫：

• 本地工具：local_echo
• 遠端工具：hub_time-get_current_time, hub_time-convert_time (自動加上前綴)

---

API 參考

WebSocketTunnelMCP

方法	說明
connect(uri)	連線到 WebSocket URI
disconnect()	斷開連線
listen_loop()	進入訊息監聽迴圈
mount_client(client)	掛載外部 MCPHttpClient

裝飾器	說明
@mcp.tool(name, description, input_schema)	註冊工具
@mcp.on_connected()	連線成功事件
@mcp.on_disconnected()	連線中斷事件

MCPHttpClient

參數	說明
base_url	MCP Server URL
client_name	客戶端名稱 (用於工具前綴)
use_sse	True = SSE 模式, False = HTTP 模式

方法	說明
start()	啟動背景連線
stop()	停止連線
call_tool(name, args)	呼叫遠端工具

---

範例程式

完整範例請參考 examples/ 目錄：

• example_basic.py - 基礎範例
• example_bridge.py - 橋接範例

---

License

MIT License - DA2 Studio