iT邦幫忙

2026 iThome 鐵人賽

DAY 27
0
AI Engineering

地端 AI 建築學系列 第 27 篇

27 案例四:Hermes Agent (3)與自建 MCP Agent 協作

  • 分享至 

  • xImage
  •  

前兩篇我們把 Hermes 的架構搞懂,也分享了怎麼幫它訓練自訂技能(Skill)。這篇要做一件更「實戰」的任務:把一個我們自己寫好、專門產生軟體規格書的 Spec Writer Agent,包裝成 MCP Server,讓 Hermes 可以直接叫它做事。

換句話說,這篇是分享怎麼讓 Hermes 不用什麼都自己會 — 遇到專業任務,就丟給旁邊那個更懂的 Agent 去處理,自己只負責溝通、調度跟記憶。


先看實驗結果:跑起來到底長怎樣

在拆架構之前,先讓大家有個畫面:這套東西接起來之後,實際執行會發生什麼事。

https://ithelp.ithome.com.tw/upload/images/20261007/20181345OMbkAGiwdk.png

流程很單純:使用者丟一句需求給 Hermes → Hermes 透過技能判斷要呼叫 generate_spec 這個 MCP 工具 → 背景的 Spec Writer Agent 開始跑六個子工具 → 中途每完成一步就回報進度 → 最後產出完整規格書 + 評分報告。完整規格書輸出參考結果如下:

https://ithelp.ithome.com.tw/upload/images/20261007/20181345ESVr04rkIP.png


協作架構:Hermes 跟自訂 Agent 怎麼分工

https://ithelp.ithome.com.tw/upload/images/20261007/201813455c0e1B0v5E.png

整張架構圖分成左右兩塊,中間用 MCP 協定串起來:

左邊:Personal AI Agent(黃)

這是使用者直接面對的那一層,角色是 Hermes(或 OpenClaw)。它做兩件事:

  1. 接收使用者的 Prompt,透過「Skill Select」判斷這個需求該用哪個技能、該呼叫哪個外部工具。

  2. 跟「Session Checkpointer」雙向溝通——把對話狀態、記憶寫成 MD 檔或存進 memory store,下次對話還能接得上。

你可以把這層想成「秘書」:它不需要自己會寫規格書、不需要自己會寫程式,它只需要知道「這種事該找誰」,然後把上下文管好。

右邊:Business AI Agent(紅)

這是真正做重活的地方。Skill Select 透過 MCP 呼叫三個各自獨立、各自專精的業務型 Agent:

  • Spec Agent —— 就是這篇的主角,負責把模糊需求變成完整規格書。

  • Coding Agent —— 負責把規格書變成實際程式碼。

  • QC Agent —— 負責品質檢驗。

三個 Agent 各自附了一個「Review Sub-agent」,代表每個業務 Agent 內部都有自己的自我審查機制(對 Spec Agent 來說,這就是後面會看到的評分-迭代迴圈)。外圍那個虛線框「Re-run Last Agent」表示:如果最後人工審核沒過,可以只重跑「最後一個」業務 Agent,不用整條 pipeline 重來。

三個業務 Agent 的產出最後都匯到 Final Review(human in the loop)——這一步是真人審核,Approve 就繼續往下走到 Final Output,Reject 則打回重做。Final Output 完成後會 Write Memory,寫回左邊的 Session Checkpointer,讓 Hermes 之後的對話還記得這次做過什麼。

這個分工設計在解決什麼問題

如果什麼都塞進一個大 Agent 裡,你會遇到幾個很實際的麻煩:System Prompt 越寫越長、工具越掛越多、Agent 常常搞不清楚現在該做哪件事、想換模型或升級某個功能還得整包一起動。

拆成「個人助理 + 一群專業業務 Agent」之後:

  • 每個業務 Agent 職責單一,可以各自獨立開發、獨立測試、獨立升版本。

  • Hermes 不需要懂 Spec Agent 內部怎麼跑 LlamaIndex,它只需要知道「有一個叫 generate_spec 的 MCP 工具可以用」。

  • 人工審核卡在 Business Agent 群組跟 Final Output 之間,形成一道品質關卡,而不是讓 AI 產出直接生效。

這篇文章要拆的,就是右邊紅框裡「Spec Agent」這一塊,是怎麼被包裝成一個 Hermes 能呼叫的 MCP 工具。


自訂 Agent 實作專案架構

spec-writer-agent/
├── spec_agent.py   # 核心
├── mcp_server.py   # 把 spec_agent 包成 MCP Server
├── mcp_client.py   # 範例 Client:呼叫 generate_spec、印出即時進度
└── outputs/                             # spec_agent 執行後自動產生的檔案
    ├── spec_v1.1_HHMMSS.md              # 第 1 輪 Spec
    ├── spec_v1.2_HHMMSS.md              # 第 2 輪(改善後)
    └── final_report_YYYYMMDD_HHMMSS.md  # 最終報告(需求+評分+完整 Spec)

hermes/skills/generate-spec/  # 放在 Hermes 的技能目錄下
└── SKILL.md  # Hermes 讀這份文件學會怎麼呼叫上面那個 MCP Server

程式碼說明

依序說明核心的 Agent 功能、MCP Server、MCP Client 與給 Hermes Agent 看的 SKILL 文件:

spec_agent.py — 核心邏輯

先看 Phase 1 怎麼確定性地跑完前五個工具:

pipeline = [
    ("normalize_requirements", {"raw_requirements": user_prompt}),
    ("generate_flowchart",     {}),
    ("generate_user_stories",  {}),
    ("initial_system_plan",    {}),
    ("integrate_spec",         {}),
]

for tool_name, kwargs in pipeline:
    try:
        _run_tool(tool_name, kwargs)
    except Exception as e:
        print(f"\n  {RED}✗  {tool_name} 執行失敗:{e}{RESET}")
        return

就是個普通的 for 迴圈依序呼叫,沒有任何 LLM 在中間做決策。接著 Phase 2 才真正交給 Agent:

react_agent = ReActAgent(
    name="spec_agent",
    description="Spec Agent",
    system_prompt=system_prompt,
    tools=tools,
    llm=llm,
    max_iterations=MAX_ITERATIONS * 5 + 2,
    timeout=3600,
)

System Prompt 裡把「什麼問題該重跑哪個工具」寫得很白話,等於是給 Agent 一份決策對照表:

【改善策略參考】
- 問題涉及「需求遺漏 / 需求不清」→ 重跑 normalize_requirements
- 問題涉及「流程不完整 / 缺少流程分支」→ 重跑 generate_flowchart
- 問題涉及「User Story 不足 / AC 不可測試」→ 重跑 generate_user_stories
- 問題涉及「技術方案不合理 / 缺少 API 或資料模型」→ 重跑 initial_system_plan
- 問題涉及「章節不齊全 / 文件結構混亂」→ 直接重跑 integrate_spec

mcp_server.py — 包成 MCP Server

這支檔案的任務只有一個:把上面那支「本機同步阻塞」的 Python 模組,變成一個 Hermes 能透過網路呼叫的 MCP 工具。

有意思的地方是這句注解點出的問題:spec_agent.py 內部的 LLM 呼叫是同步、阻塞的,如果直接在 async 的工具函式裡 await 它,整個事件迴圈會被卡死,連 progress notification 都送不出去。解法是丟到背景執行緒,中間用 queue.Queue 傳事件:

def _run_spec_agent_in_thread(requirements, event_queue, result_holder):
    def on_progress(event: dict) -> None:
        # callback 是從 spec_agent 內部的同步函式呼叫,
        # 只能做同步、輕量的事:丟進 thread-safe queue 就好。
        event_queue.put(event)

    try:
        asyncio.run(spec_agent.run_async(requirements, on_progress=on_progress))
    except Exception as e:
        result_holder["error"] = e

主協程(main coroutine)那邊則是一直輪詢這個 queue,把事件轉成 MCP 的 progress notification 送出去:

worker = threading.Thread(target=_run_spec_agent_in_thread, args=(...), daemon=True)
worker.start()

while worker.is_alive() or not event_queue.empty():
    try:
        event = event_queue.get(timeout=0.2)
    except queue.Empty:
        continue
    step += 1
    await ctx.report_progress(progress=step, message=_format_event(event))

_format_event 就是把 spec_agent 傳來的結構化事件,翻成人看得懂的一句話:

def _format_event(event: dict) -> str:
    etype = event.get("type")
    if etype == "tool_start":
        step, total = event.get("step"), event.get("total_steps")
        prefix = f"[{step}/{total}] " if step and total else ""
        return f"{prefix}開始執行:{event.get('label')}"
    if etype == "score":
        verdict = "通過" if event.get("passed") else "未通過"
        return f"評分結果:{event.get('score')}/100(第 {event.get('iteration')} 輪,{verdict})"
    ...

另外還有一個部署上實用的細節:MCP SDK 內建 DNS rebinding 防護,預設只信任 localhost,如果你要從區網其他機器連進來,得透過環境變數把允許的 host 加進白名單,不然會被擋成 421:

_allowed_hosts_env = os.environ.get("MCP_ALLOWED_HOSTS", "*")
if _allowed_hosts == ["*"]:
    _transport_security = TransportSecuritySettings(enable_dns_rebinding_protection=False)
else:
    _transport_security = TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=["localhost", "127.0.0.1", *_allowed_hosts],
        allowed_origins=["*"],
    )

mcp_client.py — 最小可行 Client

這支就是拿來驗證 Server 有沒有接好的範例,重點只有兩段:連線初始化,跟帶著 progress_callback 呼叫工具。

async def on_progress(progress: float, total: float | None, message: str | None) -> None:
    """每次收到伺服器推送的進度通知時會被呼叫。"""
    print(f"[進度 {progress:g}] {message}")

async def main(requirements: str) -> None:
    async with streamablehttp_client(MCP_SERVER_URL) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()

            result = await session.call_tool(
                "generate_spec",
                {"requirements": requirements},
                progress_callback=on_progress,
            )

            for block in result.content:
                if hasattr(block, "text"):
                    print(block.text)

這裡值得注意的是 call_tool 支援 progress_callback 這個參數 — 這就是為什麼即使 generate_spec 整個流程要跑好幾分鐘,使用端也不會呆呆盯著空白畫面,而是能即時看到「現在跑到第幾步」。Hermes 之後要接這個工具時,用的就是完全一樣的模式,只是把 print 換成它自己的訊息呈現方式。

SKILL.md — 給 Hermes 看的說明書

前面三支檔案,Hermes 完全不需要知道內部怎麼實作,它只需要讀這份 SKILL.md,就知道「有這個工具、怎麼呼叫、會踩到什麼坑」。這份文件其實才是這整套協作能不能順利運作的關鍵。

Skill 開頭的 frontmatter 定義了觸發條件:

---
name: generate-spec
description: |
  Call spec-agent-mcp-server's `generate_spec` tool to produce a full SRS document
  trigger: generate-spec, mcp, spec, srs, spec-writing, software-requirements, 規格文件
category: devops
---

小結

這篇串起來看,其實就是把架構圖裡「Skill Select → MCP → Spec Agent → Review Sub-agent」這條線,一步步變成可以實際運作的程式碼:

  • spec_agent.py 負責把 LLM 呼叫包裝成六個有明確職責的工具
  • mcp_server.py 負責把同步阻塞的本地模組變成非同步、可回報進度的網路服務
  • mcp_client.py 驗證這條線通不通
  • SKILL.md 則是讓 Hermes 學會怎麼用它

上一篇
26 案例四:Hermes Agent (2)訓練自訂技能
系列文
地端 AI 建築學 共 27 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言