da2-xiaozhi-tunnel 0.1.4
AI xiaozhi 小智 WebSocket MCP (Model Context Protocol) 橋接模組，支援 WebSocket 與 HTTP/SSE 傳輸
License: MIT
Requires Python: >=3.10
Keywords: mcp, websocket, sse, xiaozhi, ai, bridge, da2, xiaozhi-tunnel, mcp-proxy, mcp-pipe

# da2-xiaozhi-tunnel

[![PyPI version](https://badge.fury.io/py/da2-xiaozhi-tunnel.svg)](https://badge.fury.io/py/da2-xiaozhi-tunnel)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

MCP (Model Context Protocol) 橋接模組，用於連接小智雲端與其他 MCP 伺服器。

**作者**: Danny | **工作室**: DA2 Studio | **網站**: [https://da2.35g.tw](https://da2.35g.tw)

---

## 功能特色

- **WebSocketTunnelMCP**: WebSocket 隧道伺服器，對接小智雲端
- **MCPHttpClient**: HTTP/SSE 客戶端，連接遠端 MCP 伺服器
- 支援多伺服器工具聚合與智慧路由
- 自動重連機制

---

## 安裝

```bash
pip install da2-xiaozhi-tunnel
```

---

## 快速開始

### 基礎範例 (僅使用 WebSocketTunnelMCP)

連接小智雲端，註冊本地工具供 AI 呼叫：

```python
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，聚合所有工具：

```python
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/`](examples/) 目錄：
- `example_basic.py` - 基礎範例
- `example_bridge.py` - 橋接範例

---

## License

MIT License - DA2 Studio