今天的目標與挑戰

到目前我已經封裝好 WhisperService，並在 Docker 容器中啟動了 n8n 並建立了 Webhook。今天的任務，是將 WhisperService 與 n8n Webhook 串接起來，讓 MCP Agent 完成從「音訊 → Whisper 轉錄 → Webhook 觸發」的流程。

• 撰寫 MCPAgent 類別，並且整合 WhisperService
• 在 MCP Agent 中呼叫 n8n Webhook，傳送轉錄結果
• 撰寫並執行端到端測試程式
• 驗證整合後的完整流程

Step 1：建立 MCPAgent 類別

先在 src/ 資料夾下新增 mcp_agent.py，並撰寫 MCPAgent 類別骨架

import requests
from src.whisper_service import WhisperService

class MCPAgent:
    def __init__(self, model="medium",
                 webhook_url="http://localhost:5678/webhook/m2a-test"):
        self.whisper = WhisperService(model)
        self.webhook_url = webhook_url

    def process_audio(self, audio_path, instruction):
        pass

儲存後，先確認檔案可正常匯入，且無語法錯誤就好。

Step 2：整合 WhisperService 轉錄

在 process_audio 方法中，先呼叫 WhisperService 進行轉錄，並檢查是否有錯誤

def process_audio(self, audio_path, instruction):
    # 1. 呼叫 WhisperService 轉錄
    result = self.whisper.transcribe(audio_path)
    if result.get("error"):
        return {"error": f"轉錄失敗：{result['error']}"}

    # 2. 準備要送給 Webhook 的 payload
    payload = {
        "text": result["text"],
        "instruction": instruction
    }

儲存後，在 Python 互動式終端機測試轉錄

python
>>> from src.mcp_agent import MCPAgent
>>> agent = MCPAgent()
>>> result = agent.whisper.transcribe("recording/小妹妹介紹她的玩偶.m4a")
>>> print(result)

執行結果

會印出一個字典，裡面包含 text、segments、duration、transcribe_time，且 error 為 None，這樣就表示轉錄成功。

Step 3：呼叫 n8n Webhook

接著在 process_meeting 裡新增對 Webhook 的 POST 請求，並且回傳最終結果

        # 3. 呼叫 n8n Webhook
        try:
            resp = requests.post(self.webhook_url, json=payload)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            return {"error": f"Webhook 呼叫失敗：{e}"}

完整的 process_meeting 函數如下

    def process_audio(self, audio_path, instruction):
        result = self.whisper.transcribe(audio_path)
        if result.get("error"):
            return {"error": f"轉錄失敗：{result['error']}"}

        payload = {"text": result["text"], "instruction": instruction}

        try:
            resp = requests.post(self.webhook_url, json=payload)
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            return {"error": f"Webhook 呼叫失敗：{e}"}

儲存後，一樣在 Python 互動式終端機測試轉錄

python
>>> from src.mcp_agent import MCPAgent
>>> agent = MCPAgent()
>>> result = agent.whisper.transcribe("recording/小妹妹介紹她的玩偶.m4a")
>>> print(result)

執行結果

在 Python 互動式環境中導入 MCPAgent 並執行 process_meeting，成功回傳 {'status':'ok'}，代表 Webhook 收到並且正確回應。

Step 4：撰寫端到端測試程式

在專案根目錄新增 test_mcp_agent.py 驗證完整的流程

from src.mcp_agent import MCPAgent

def main():
    agent = MCPAgent(
        model="medium",
        webhook_url="http://localhost:5678/webhook/m2a-test"
    )
    result = agent.process_meeting(
        "recording/小妹妹介紹她的玩偶.m4a",
        "請幫我整理出摘要"
    )
    print("最終結果：", result)

if __name__ == "__main__":
    main()

執行

python test_mcp_agent.py

執行結果

最終結果： {'status':'ok'}

整個「音訊 → Whisper 轉錄 → Webhook 觸發 → 回應」流程順利完成，結果符合我們的預期！

今天的成果總結

✅ 完成項目

• 撰寫 MCPAgent 類別，整合 WhisperService
• 在 MCP Agent 中實作對 n8n Webhook 的呼叫
• 撰寫並執行端到端測試，成功回傳 {'status':'ok'}
• 驗證了從音訊檔到工作流觸發的完整流程

📝 關鍵發現

• 服務化設計：讓核心流程變得簡單明瞭，只需要一個方法就能完成所有動作
• HTTP 通訊迅速：只要網路通暢，Webhook 呼叫幾乎即時回應
• 可重複測試：每次執行都能回傳相同格式結果，方便後續擴展與調試

心得

今天將我目前的流程串接至 n8n 工作流成功，讓我感到興奮，明天繼續往下一步邁進！

🎯 明天計劃
研究如何使用部署本地 AI 模型，並將其 API 整合進 n8n 工作流中，以實現免費、離線且可靠的繁體中文摘要功能。