打造 AI 工具橋接器：da2-xiaozhi-tunnel 深度解析

讓你的小智 AI 擁有無限可能的工具擴展能力

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

---

前言：為什麼需要 MCP Tunnel？

在使用小智 AI 的過程中，你是否曾經想過：

• 「如果小智能幫我查詢本地資料庫就好了...」
• 「要是能讓 AI 控制我的智慧家電多好...」
• 「能不能把其他 MCP Server 的工具也整合進來？」

da2-xiaozhi-tunnel 就是為了解決這些問題而誕生的！

它是一個 MCP (Model Context Protocol) 橋接模組，讓你能夠：

1. 將自訂工具註冊給小智 AI 使用
2. 橋接其他 MCP Server，聚合所有工具
3. 讓 AI 透過 WebSocket 即時呼叫你的程式

---

核心概念：什麼是 MCP？

MCP (Model Context Protocol) 是一個讓 AI 模型與外部工具溝通的標準協定。

想像一下： - AI 是一個聰明的大腦 - 工具 (Tools) 是各種功能的手腳 - MCP 就是連接大腦與手腳的神經系統

┌─────────────────────────────────────────────────────────────┐
│                        小智雲端 AI                           │
│                   (理解語意、決策呼叫)                        │
└─────────────────────────────┬───────────────────────────────┘
                              │ WebSocket (MCP 協定)
                              ▼
┌─────────────────────────────────────────────────────────────┐
│              da2-xiaozhi-tunnel (你的電腦)                    │
│  ┌───────────────────┐  ┌────────────────────────────────┐  │
│  │   本地工具         │  │       外部 MCP Server          │  │
│  │  - 查天氣         │  │  - 資料庫查詢                  │  │
│  │  - 開燈           │  │  - 智慧家電控制                │  │
│  │  - 計算機         │  │  - 第三方 API                  │  │
│  └───────────────────┘  └────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

---

安裝

一行指令搞定：

pip install da2-xiaozhi-tunnel

---

快速開始：基礎範例

讓我們從最簡單的範例開始 — 註冊一個本地工具給小智使用。

步驟一：取得 Token URL

1. 登入 小智雲端平台
2. 進入 MCP 設定頁面
3. 複製你的 Token URL (格式：wss://api.xiaozhi.me/mcp/?token=xxx)

步驟二：撰寫程式碼

"""
example_basic.py - 基礎範例 (僅使用 WebSocketTunnelMCP)

此範例展示如何使用 WebSocketTunnelMCP 連接小智雲端，
並註冊本地工具供 AI 呼叫。

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

import asyncio
from da2_xiaozhi_tunnel import WebSocketTunnelMCP

# ================== 設定區 ==================
# ★ 請填入您的 Token URL ★
TOKEN_URL = "wss://api.xiaozhi.me/mcp/?token=YOUR_TOKEN_HERE"
# ============================================

# 建立 MCP 伺服器實例
mcp = WebSocketTunnelMCP(server_name="BasicDemo")

# ===== 連線事件 =====
@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}！歡迎使用 DA2 MCP Tunnel！"

@mcp.tool(
    name="calculate",
    description="執行簡單的數學計算",
    input_schema={
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "數學表達式，例如 '1 + 2 * 3'"
            }
        },
        "required": ["expression"]
    }
)
async def calculate(expression: str):
    """計算數學表達式"""
    try:
        # 注意：實際生產環境應使用更安全的方式
        result = eval(expression)
        return f"計算結果: {expression} = {result}"
    except Exception as e:
        return f"計算錯誤: {e}"

# ===== 主程式 =====

async def main():
    print("=" * 50)
    print(" DA2 MCP Tunnel - 基礎範例")
    print(" 按 Ctrl+C 結束程式")
    print("=" * 50)

    while True:
        try:
            print(f"正在連線: {TOKEN_URL[:60]}...")
            if await mcp.connect(TOKEN_URL):
                await mcp.listen_loop()
        except KeyboardInterrupt:
            print("\n程式結束")
            break
        except Exception as e:
            print(f"連線錯誤: {e}")

        print("5 秒後重新連線...")
        await asyncio.sleep(5)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n程式結束")

步驟三：執行程式

python example_basic.py

執行後你會看到：

==================================================
 DA2 MCP Tunnel - 基礎範例
 按 Ctrl+C 結束程式
==================================================
正在連線: wss://api.xiaozhi.me/mcp/?token=xxx...
已連線至小智伺服器

現在你可以對小智說： - 「請跟 Danny 打個招呼」→ AI 會呼叫 get_greeting("Danny") - 「幫我算 123 * 456」→ AI 會呼叫 calculate("123 * 456")

---

深入解析：程式碼結構

建立 MCP 實例

mcp = WebSocketTunnelMCP(server_name="BasicDemo")

參數	說明
server_name	伺服器名稱，會在 MCP 握手時回傳給小智
logger	(可選) 自訂日誌函式，預設為 print

註冊事件處理器

@mcp.on_connected()
async def on_connected():
    print("已連線")

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

這兩個裝飾器讓你在連線/斷線時執行自訂邏輯，例如： - 連線成功時初始化資源 - 斷線時清理暫存資料

註冊工具 (@mcp.tool)

這是最重要的部分！工具定義包含三個要素：

@mcp.tool(
    name="get_greeting",           # 工具名稱 (AI 呼叫時使用)
    description="取得問候語",        # 工具描述 (讓 AI 理解用途)
    input_schema={...}             # 參數定義 (JSON Schema 格式)
)
async def get_greeting(name: str):
    return f"哈囉 {name}！"

input_schema 使用 JSON Schema 格式，告訴 AI： - 這個工具需要什麼參數 - 每個參數是什麼型別 - 哪些參數是必填的

主程式迴圈

while True:
    if await mcp.connect(TOKEN_URL):
        await mcp.listen_loop()
    await asyncio.sleep(5)  # 斷線後等待 5 秒重連

這個結構提供了自動重連機制： - connect() 建立 WebSocket 連線 - listen_loop() 持續監聽訊息，直到斷線 - 斷線後等待 5 秒再重試

---

進階應用：橋接外部 MCP Server

基礎範例只用到本地工具，但 da2-xiaozhi-tunnel 的真正威力在於 橋接功能！

你可以掛載多個外部 MCP Server，將它們的工具聚合後一起提供給小智。

架構圖

                    ┌─────────────────────────────────────┐
                    │         小智雲端 (Xiaozhi)           │
                    │          (WebSocket)                │
                    └─────────────┬───────────────────────┘
                                  │
                    ┌─────────────▼───────────────────────┐
                    │      WebSocketTunnelMCP (Host)       │
                    │  ┌─────────────────────────────┐    │
                    │  │  本地工具 (Local Tools)       │    │
                    │  │  - local_echo               │    │
                    │  │  - get_system_info          │    │
                    │  └─────────────────────────────┘    │
                    │  ┌─────────────────────────────┐    │
                    │  │  外部工具 (External Tools)    │    │
                    │  │  - hub_time-get_current_time │    │
                    │  │  - hub_db-query             │    │
                    │  └─────────────────────────────┘    │
                    └──────────────┬──────────────────────┘
                                   │
               ┌───────────────────┴───────────────────┐
               │                                       │
    ┌──────────▼──────────┐             ┌──────────────▼──────────┐
    │  MCPHttpClient #1   │             │   MCPHttpClient #2      │
    │  (client_name: hub) │             │   (client_name: db)     │
    │  HTTP/SSE 模式       │             │   HTTP/SSE 模式          │
    └──────────┬──────────┘             └──────────────┬──────────┘
               │                                       │
    ┌──────────▼──────────┐             ┌──────────────▼──────────┐
    │  遠端 MCP Server #1  │             │   遠端 MCP Server #2    │
    │  (mcphub.io)        │             │   (內部 API)             │
    └─────────────────────┘             └─────────────────────────┘

橋接範例程式碼

"""
example_bridge.py - 橋接範例 (WebSocketTunnelMCP + MCPHttpClient)

此範例展示如何使用 WebSocketTunnelMCP 連接小智雲端，
同時橋接遠端 MCP Server，將其工具聚合後一起提供給 AI。

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

import asyncio
from da2_xiaozhi_tunnel import WebSocketTunnelMCP, MCPHttpClient

# ================== 設定區 ==================
# ★ 請填入您的 Token URL ★
TOKEN_URL = "wss://api.xiaozhi.me/mcp/?token=YOUR_TOKEN_HERE"

# ★ 外部 MCP Server 位址 ★
EXTERNAL_MCP_URL = "http://192.168.10.56:30050/mcp"
# ============================================

# 建立 MCP 伺服器實例
mcp = WebSocketTunnelMCP(server_name="BridgeDemo")

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

# ===== 連線事件 =====
@mcp.on_connected()
async def on_connected():
    print("已連線至小智伺服器")
    print(f"本地工具數量: {len(mcp.tools)}")
    print(f"外部工具數量: {len(mcp.external_tools)}")

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

# ===== 註冊本地工具 =====

@mcp.tool(
    name="local_echo",
    description="本地回聲工具 - 回傳收到的訊息",
    input_schema={
        "type": "object",
        "properties": {
            "message": {
                "type": "string",
                "description": "要回聲的訊息"
            }
        },
        "required": ["message"]
    }
)
async def local_echo(message: str):
    """回聲工具：回傳收到的訊息"""
    return f"[本地回聲] {message}"

@mcp.tool(
    name="get_system_info",
    description="取得系統資訊",
    input_schema={"type": "object", "properties": {}}
)
async def get_system_info():
    """取得系統資訊"""
    import platform
    return f"系統: {platform.system()} {platform.release()}, Python: {platform.python_version()}"

# ===== 主程式 =====

async def main():
    print("=" * 60)
    print(" DA2 MCP Tunnel - 橋接範例 (Bridge Demo)")
    print(" 本地工具 + 遠端 MCP Server 工具聚合")
    print(" 按 Ctrl+C 結束程式")
    print("=" * 60)

    # 掛載外部 Client (需在 event loop 內執行)
    print(f"掛載外部 MCP: {EXTERNAL_MCP_URL}")
    mcp.mount_client(external_client)

    while True:
        try:
            print(f"正在連線: {TOKEN_URL[:60]}...")
            if await mcp.connect(TOKEN_URL):
                await mcp.listen_loop()
        except KeyboardInterrupt:
            print("\n程式結束")
            await external_client.stop()
            break
        except Exception as e:
            print(f"連線錯誤: {e}")

        print("5 秒後重新連線...")
        await asyncio.sleep(5)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n👋 程式結束")

關鍵差異

功能	基礎範例	橋接範例
只用本地工具	V	V
掛載外部 MCP	X	V
工具來源	僅本地	本地 + 遠端

---

MCPHttpClient 詳解

MCPHttpClient 是用來連接外部 MCP Server 的客戶端，支援兩種模式：

HTTP 模式 (推薦)

client = MCPHttpClient(
    base_url="http://your-server/mcp",
    client_name="hub",
    use_sse=False  # 純 HTTP 模式
)

連線流程： 1. POST /mcp → initialize 請求 2. 從 Response Header 擷取 mcp-session-id 3. POST /mcp → tools/list 4. 完成！工具已可用

SSE 模式

client = MCPHttpClient(
    base_url="http://your-server/mcp",
    client_name="hub",
    use_sse=True  # SSE 模式
)

連線流程： 1. GET /sse → 建立 SSE 長連線 2. 等待 endpoint 事件 3. POST → initialize / tools/list 4. 持續監聽 SSE 事件

工具前綴命名

當你掛載一個 client_name="hub" 的 Client，其工具會自動加上前綴：

• 原始工具名：get_time
• 聚合後名稱：hub_get_time

這樣可以避免多個 Server 的工具名稱衝突！

---

智慧路由機制

當 AI 呼叫工具時，WebSocketTunnelMCP 會自動判斷該呼叫哪裡：

收到 tools/call 請求
        │
        ▼
┌───────────────────┐
│ 工具名在本地嗎？   │ ─── Yes ──→ 執行本地函式
└───────────────────┘
        │ No
        ▼
┌───────────────────┐
│ 匹配外部 Client？  │ ─── Yes ──→ 轉發給 MCPHttpClient
│ (檢查前綴)        │
└───────────────────┘
        │ No
        ▼
    回傳 "Tool not found"

---

常見問題 FAQ

Q: 連線一直失敗怎麼辦？

A: 請檢查： 1. Token URL 是否正確 2. 網路是否能訪問 api.xiaozhi.me 3. Token 是否過期

Q: 外部 MCP Server 連不上？

A: 確認： 1. Server URL 是否正確 2. Server 是否支援 HTTP/SSE 模式 3. 嘗試切換 use_sse 參數

Q: 工具沒有出現在小智？

A: 檢查： 1. 確認 @mcp.tool 裝飾器語法正確 2. 確認 input_schema 為合法 JSON Schema 3. 重新連線試試

---

結語

da2-xiaozhi-tunnel 讓你輕鬆擴展小智 AI 的能力，無論是本地工具還是遠端服務都能整合。

現在就開始打造屬於你的 AI 工具吧！🚀

---

相關資源

• PyPI: https://pypi.org/project/da2-xiaozhi-tunnel/
• GitHub: 完整範例請參考 examples/ 目錄
• DA2 Studio: https://da2.35g.tw