標籤: API

  • A2A 是什麼:用一個 Python 把資料庫變成 AI 外掛技能

    重點摘要

    • A2A 是什麼:把服務包成「一張 AI 看得懂的名片(Agent Card)+ 一個任務入口(/tasks)」,別的 AI 給它一個網址就能自己使用你。
    • 跟 MCP 的差別:MCP 是把「死工具」掛給一個 AI(像 JDBC 驅動);A2A 是讓兩個「會思考的服務」互相喊話(像微服務互呼)。
    • 怎麼被呼叫:分兩層——LLM 讀名片自己挑 skill 是「自然語言層」;身分憑證走 HTTP header 是「水管層」,兩層不混。
    • 只有一個 DB:本質就是「一個 Flask + 一張名片」,本文附可跑程式碼。
    • 權限怎麼帶:token 不放 skill 參數、放 header;「能看什麼」由 server 從 token 推導,不讓呼叫方自報。認證掛在 Nginx 也能發名片——把發現路徑設成白名單即可。

    這篇從「A2A 是什麼」一路講到「放上 production、要管權限時怎麼設計」。對象是上一個世代的後端 RD——你熟悉 REST、微服務、JDBC、Nginx、服務發現,所以全程用你已經會的概念來對照。最後的程式碼我自己在跑,而且真的被另一台機器的 agent 呼叫成功過。

    一、A2A 是什麼?

    A2A(Agent2Agent)是一個讓「AI agent 之間互相呼叫」的開放協議。它由 Google 提出,現已捐給 Linux Foundation,版本到 v1.0。核心只有兩個東西:

    1. Agent Card(名片):一份放在固定網址的 JSON,描述「我是誰、我會哪些技能(skills)、任務要打哪個網址、要怎麼認證」。別的 agent 靠它「發現」你。
    2. Task 入口:一個收任務的端點(例如 POST /tasks),別人帶著「我要用哪個 skill」打進來,你執行、回 JSON。

    就這樣。一個 A2A agent,對別的 AI 來說,就像一個「外掛技能」:本來不會查你系統的 LLM,你給它一個網址,它讀了名片就「學會」用你了——不需要工程師事先把兩邊接線。

    用生活化比喻:名片 + 通訊錄

    想像你是一間公司的窗口。你印了一張名片,上面寫「我負責:查包裹、開通知單、查住戶數」。任何人拿到這張名片,看一眼就知道能找你做什麼、該打哪支電話。A2A 的 Agent Card 就是這張名片;/tasks 就是那支電話。差別只在於:讀名片、打電話的不是人,是另一個 AI。

    二、上世代 RD 怎麼理解 A2A 與 MCP 的差別?

    這是最多人卡住的地方。用你熟悉的後端概念對照,一秒就懂:

    你熟悉的概念 MCP A2A
    JDBC / ODBC 驅動程式 ✓ 標準介面,把工具(DB/檔案/API)接給「一個」AI 用
    微服務互相呼叫(REST) ✓ 一個服務呼叫「另一個會思考的服務」
    服務發現(Eureka / Consul) ✓ 靠 Agent Card 自我描述被發現
    對方是「活的」還是「死的」 死的:工具被動被呼叫,不會自己判斷 活的:對方是個會用 LLM 自己判斷的 agent

    一句話收斂:MCP 是「AI ↔ 工具」,A2A 是「AI ↔ AI」。MCP 把一隻手伸向工具箱;A2A 是兩個有腦袋的人互相打電話。兩者不衝突,常常一起用——你的 agent 用 MCP 拿工具,同時用 A2A 去問別的 agent。

    三、A2A 怎麼跟 AI Agent 一起運用?(MCP 與 Skill 的分工)

    • MCP:當你的 AI 需要用「工具」——查 DB、讀檔案、呼叫某個 API——就掛一個 MCP server。工具不會思考。
    • A2A:當你的 AI 需要請「另一個 agent」幫忙——對方有自己的 domain 知識、會自己判斷——就用 A2A 呼叫它。
    • Skill:在 A2A 名片裡,一個 agent 把自己會的事拆成一個個「技能(skill)」宣告出來。呼叫方的 LLM 讀這些技能描述,自己挑要用哪一個。

    所以 Skill 是 A2A 名片裡的最小單位。一張名片可以宣告多個 skill,每個 skill 背後接不同的後端——有的查即時 DB、有的走知識圖譜、有的呼叫生成式模型。呼叫方不需要知道背後怎麼接,它只看得到「技能清單」。

    四、怎麼用自然語言「呼叫」一個 A2A agent?

    關鍵觀念:呼叫方不寫死「呼叫 skill X」。它把「名片上的技能清單」加上「使用者講的自然語言」一起交給 LLM,讓 LLM 自己挑出最適合的 skill,再去打 /tasks。流程三步:發現 → 判斷 → 呼叫。這是「判斷」那一步的真實程式碼:

    def llm_pick(user_says, skills):
        menu = "\n".join(f"- {s['id']}: {s['name']} — {s['description']}" for s in skills)
        prompt = (f"使用者說:「{user_says}」\n"
                  f"下面是某個 agent 提供的 skill,挑最適合處理這句話的一個,"
                  f"只回那個 skill 的 id、不要任何解釋:\n{menu}")
        # ...把 prompt 丟給 LLM,回應裡比對出 skill id

    先記住這個分層:自然語言層 vs 水管層(後面講權限會用到)

    一次 A2A 呼叫其實有兩層,先分清楚,等一下講權限才不會打結:

    誰在做 放什麼
    自然語言層 LLM 讀名片、挑 skill 業務參數(query、id)
    水管層(HTTP) client 程式發請求 身分憑證(Authorization header)

    所以「用自然語言達到 A2A」,是把對方的名片餵給你的 LLM、讓它把人話翻成「該呼叫哪個 skill」;而身分憑證從頭到尾不經過 LLM,是底層水管自動帶的。名片寫得好,LLM 就挑得準。

    五、只有一個 DB,怎麼做出一個 A2A agent?

    直接回答最多人的疑問:對,本質就是「一個 Python 寫的 API 口」——只是比普通 REST API 多兩樣:一張名片、用 skill 分路。下面是能跑的最小骨架(Flask),背後接一個資料庫:

    from flask import Flask, request, jsonify
    app = Flask(__name__)
    
    # ① 名片:宣告我會哪些 skill、task 打哪
    CARD = {
      "name": "社區管理大腦",
      "url": "http://localhost:9999",
      "skills": [
        {"id": "parcel_status", "name": "包裹卡單查詢",
         "description": "查 outbound 各狀態的真實件數,即時查 DB"},
      ],
    }
    
    @app.get("/.well-known/agent-card.json")   # 別人靠這個網址發現我
    def card(): return jsonify(CARD)
    
    @app.post("/tasks")                          # 別人帶 skill 打進來
    def tasks():
        skill = request.get_json().get("skill")
        if skill == "parcel_status":
            rows = db_query("SELECT status, COUNT(*) FROM parcels "
                            "WHERE direction='outbound' GROUP BY status")
            return jsonify({"facts": rows})
        return jsonify({"error": "unknown skill"}), 400

    名片 endpoint + task endpoint + 一句 SQL,你的資料庫就成了一個 A2A agent。要加技能,就在 CARD["skills"] 多宣告一筆、在 /tasks 多一條分支。

    把名片從「自我介紹」升級成「可被機器編排的技能契約」

    上面那張名片只夠「人」看懂。要讓另一個 AI 自動挑對 skill、帶對參數,再補三個欄位:parameters(要帶什麼)、returns(會拿到什麼)、whenToUse(什麼情境該挑我)。這等於「透過 API 發布一份技能契約」:

    {
      "name": "Tom 的部落格 Agent",
      "usageHint": "每個 skill 附 parameters/returns/whenToUse,呼叫方 LLM 讀完即可自行編排,無需接線。",
      "skills": [
        {
          "id": "search_posts",
          "name": "用關鍵字搜尋文章",
          "description": "全文搜尋,回最相關的前 5 篇(標題/摘要/連結/id)",
          "parameters": { "query": {"type": "string", "required": true, "desc": "搜尋關鍵字"} },
          "returns":    "hits[]{id,title,link,date,excerpt} + total",
          "whenToUse":  "使用者問題能抽出明確關鍵字、要定位文章時"
        }
      ]
    }

    有了 whenToUse + parameters,呼叫方的 LLM 就能自己編排多步。這就是「用 Data + API 把你的資料融入 AI」的具體長相:資料是真實內容,API 是入口,而這張契約是讓 AI 自己會用的關鍵。

    它真的被呼叫成功了——一段真實的 server log

    把這個 agent 跑起來後,區網另一台機器(192.168.0.51)的一個 agent 來呼叫,log 完整記錄了它「猜協議 → 發現名片 → 從錯誤學會 → 成功」的過程:

    # 它先亂猜各種常見慣例(全 404):
    GET /health ... /v1/chat/completions ... /mcp ... /openapi.json   → 404
    
    # 命中標準路徑:
    GET  /.well-known/agent-card.json   → 200   ← 讀到名片!
    POST /tasks                         → 400   ← 沒給對 skill,被回 available 清單
    POST /tasks                         → 200   ← 從錯誤學到 skill 名,改對了,成功

    注意那個 400 → 200:呼叫方第一次打錯,server 回了「可用技能清單」,它自己讀懂、改對、成功。這就是 A2A 的精神——服務自我描述,呼叫方自己學會用,中間沒有工程師接線。

    六、同一個 /tasks 入口,不同 skill 走不同後端

    對外永遠是同一個 /tasks 入口,進來後依 skill 分流。下面四個 handler 各代表一種典型後端——呼叫生成式 Agentic SDK、走快取、查 DB、純快速計算——呼叫方完全不需要知道背後差異。

    import anyio
    from claude_agent_sdk import query, AssistantMessage, TextBlock
    
    # ① Agentic SDK 後端:呼叫會思考的 agent 生成文字(慢、秒級)
    def route_draft_notice(body):
        prompt = f"用繁體中文寫一段 50 字內催領通知,目前有 {body.get('stuck', 0)} 件待領包裹。"
        async def _ask():
            out = []
            async for msg in query(prompt=prompt):          # Claude Agent SDK 一次性查詢
                if isinstance(msg, AssistantMessage):
                    out += [b.text for b in msg.content if isinstance(b, TextBlock)]
            return "".join(out)
        return {"route": "agentic_sdk", "generated": anyio.run(_ask)}
    
    # ② 快取後端:第一次 miss 才查 DB,之後走記憶體(毫秒)
    def route_community_stats(body):
        hit = cache_get("community_stats")
        if hit is not None:
            return {"route": "cache_hit", "data": hit}
        data = {"households": db_query("SELECT COUNT(*) FROM households")[0][0]}
        cache_set("community_stats", data)
        return {"route": "cache_miss_then_db", "data": data}
    
    # ③ DB 後端:即時查真實事實,完全不靠模型
    def route_parcel_status(body):
        rows = db_query("SELECT status, COUNT(*) FROM parcels "
                        "WHERE direction='outbound' GROUP BY status")
        return {"route": "db_realtime",
                "facts": [{"status": s, "count": c} for s, c in rows]}
    
    # ④ 快速計算後端:純算、無 IO(最快)
    def route_late_fee(body):
        days = int(body.get("overdue_days", 0))
        return {"route": "compute", "overdue_days": days, "late_fee": min(days, 30) * 5}

    /tasks 本身只是一張「skill → handler」對照表,進來分流出去就好。名片對外宣告技能,但背後是 DB、快取、純算還是會思考的 agent,全藏在門面後;加後端只要多一個 handler + 對照表一行。

    七、權限:誰能看什麼資料?(認證 vs 授權)

    真實系統的 DB 一定有「誰能看哪些 row」。新手最常問:這種權限要不要做成 skill 參數、讓使用者連的時候把帳密帶進去?都不要。先把兩件事拆開:

    • 認證(Authentication)= 你是誰 → 驗 token 簽章
    • 授權(Authorization)= 你能看哪些 row → server 從已驗證的身分推導範圍

    核心原則一句話:token 不放 skill 參數、放 HTTP header;「能看什麼」由 server 從 token 推導,不讓呼叫方自報。還記得第四段的兩層嗎——身分屬於「水管層」,不屬於「自然語言層」。A2A 名片用 securitySchemes 宣告認證方式(Bearer JWT / API Key / OpenID Connect),呼叫方把憑證放 Authorization header,而不是塞進 skill 的 parameters

    # 名片宣告認證方式(A2A securitySchemes)
    CARD["securitySchemes"] = {"bearer": {"type": "http", "scheme": "Bearer", "bearerFormat": "JWT"}}
    CARD["security"] = [{"bearer": []}]
    
    @app.post("/tasks")
    def tasks():
        # ① 認證:token 從 HEADER 取(不是 body 參數!),驗簽
        token = request.headers.get("Authorization", "").removeprefix("Bearer ").strip()
        claims = verify_jwt(token)
        if not claims:
            return jsonify({"error": "unauthorized"}), 401
        ctx = {"user_id": claims["sub"], "community_id": claims["community_id"], "role": claims["role"]}
    
        body = request.get_json()
        handler = SKILL_MAP.get(body.get("skill"))
        return jsonify(handler(body, ctx))         # ② 把已驗證身分傳給 handler
    
    # ③ 授權:範圍來自 ctx(token),不是 body —— caller 改不了別人的 community_id
    def route_parcel_status(body, ctx):
        rows = db_query("SELECT status, COUNT(*) FROM parcels "
                        "WHERE community_id = %s GROUP BY status", ctx["community_id"])
        return {"facts": rows}

    更保險的做法是把範圍下推到 DB 層的 RLS(Row-Level Security):handler 只從 ctx 設一次,之後資料庫自己擋,handler 想繞都繞不過。為什麼一定要這樣?因為 caller 帶進來的 body 一律不可信——它可以亂填 community_id=別人的。讓 caller 自己指定範圍,等於誰來問都能撈全部,這就是經典的 confused deputy(被搞混的代理人) 漏洞。

    那個 token 哪來?放哪?(你不用把帳密唸給 AI 聽)

    token 不走自然語言。使用者對登入端登入一次、換到一個短期 JWT,之後由 client 程式自動帶,LLM 碰都碰不到。token 放在一個秘密檔(像 .env,chmod 600),呼叫時讀出來貼 header——不貼進對話、也不做成 skill:

    # token 放秘密檔,呼叫時才讀出來貼 header(整個過程沒人看到 token 的值)
    curl -H "Authorization: Bearer $(cat ~/.config/a2a/token)" \
         -d '{"skill":"parcel_status"}' https://your-agent/tasks
    
    # 帳密要定期刷 token 的話:先拿憑證換動態 token,再打 A2A
    TOKEN=$(curl -s -u "$CLIENT_ID:$CLIENT_SECRET" https://idp/oauth/token | jq -r .access_token)
    curl -H "Authorization: Bearer $TOKEN" -d '{"skill":"parcel_status"}' https://your-agent/tasks

    定期刷的場景就是 OAuth2 的 client-credentials / refresh 流程:你準備好「帳密 + 一個換 token 的小程式」,呼叫方先拿帳密換動態 token,再打 A2A,過期就再換一次。注意:發 token 的(登入端 / IdP)要跟驗 token 的(A2A server)分開,別把登入服務塞進 A2A 本身。

    八、發現 vs 認證:服務全鎖時,名片還怎麼發?

    這裡有個雞生蛋問題:如果服務要認證,那「還沒有帳密的人」怎麼讀得到名片、知道怎麼用?更現實的是——大多數系統的 auth 掛在最前面的 Nginx 層、ingress annotation 或 WAF,請求根本到不了你的 router 就被踢掉,那名片不也一起被擋了?

    解法一:名片分兩層,公開的那層本來就免認證

    名片 路徑 要認證? 內容
    公開名片 /.well-known/agent-card.json ❌ 免 基本技能 + 「我怎麼認證」(securitySchemes)
    進階名片 /extendedAgentCard ✅ 要 驗證後才看得到的完整 / 隱藏技能

    「怎麼用我」這件事本身是公開的:公開名片不需要帳密就讀得到,而且它自己會告訴你「要進來請走這個認證方式」。發現永遠在認證之前。至於最初那組帳密哪來?A2A 協議不發帳密——它是透過 out-of-band(協議之外)拿到的,也就是一個人的決定:管理員幫你開帳號、給你 client_id/secret。協議不會從零變出信任。

    解法二:把發現路徑設成 Nginx 白名單(跟 /login 同一招)

    你早就有一批「必須公開的 bootstrap 路徑」:/healthz(LB 探活)、/.well-known/acme-challenge/*(Let’s Encrypt 續憑證)、/oauth/token(認證端自己不能要認證)。公開 Agent Card 就是再加一條進這個白名單而已:

    # 其他全鎖,只開這一個發現路徑
    location = /.well-known/agent-card.json {
        auth_request off;          # 不套那道 auth
        proxy_pass http://a2a_backend;
    }
    location / {
        auth_request /_auth;       # /tasks、/extendedAgentCard 照常鎖
        proxy_pass http://a2a_backend;
    }

    安全上不虧:A2A spec 明講公開名片不可放機密,它只放「名稱 + 怎麼認證 + 非敏感技能」,敏感技能放進階名片。所以開白名單不洩漏東西,只是貼一張「敲門方式」在門口。

    解法三:全內網死鎖時,名片根本不從這台機器送

    如果你連白名單都不想開(整個服務鎖在防火牆後),那就不要用公開發現,改用 A2A 另外兩種發現方式。關鍵觀念:名片只是一份可攜的 JSON,不一定要從「被保護的那台機器」送出去——發現 ≠ 執行,名片可以跟服務不同台。

    • Direct Configuration:你直接把名片 JSON(或內部 URL)手動給授權的呼叫方。
    • 內部 Registry:名片發布到一個內部 agent 目錄,授權 client 去那查。

    呼叫方本來就得在你的網路邊界內(VPN / 內網)才打得到 /tasks,那它也能從同一個內網管道拿到名片——你不用為了發名片在防火牆上戳洞

    九、實戰會踩的坑

    • a2a-sdk 版本坑:網路範例多半用 A2AStarletteApplication(0.x,pydantic + Starlette);但現在 pip install a2a-sdk 裝到的 1.1.0 是 protobuf / gRPC 生成,API 完全不同,照舊範例會整段跑不起來。協議穩定,SDK 的 API 會變——手刻一個極簡 HTTP server 反而最穩。
    • 把認證塞進 skill 參數:最常見的設計錯。憑證走 header、走 securitySchemes,絕不放進 skill 的 parameters;範圍永遠由 server 從 token 推導。
    • Cloudflare / WAF 擋 UA:agent 去抓套了 Cloudflare 的服務,Python urllib 不帶 User-Agent 會被回 403;帶一個瀏覽器 UA 就 200。

    結論:A2A = API 口 + 一張名片 + 一道標準的認證

    把 A2A 想成「微服務互呼,但對方是會思考的 agent,用一張名片自我介紹、讓呼叫方的 LLM 自己學會用」,核心就抓到了。技術上它可以小到一個 Flask 檔 + 一句 SQL;放上 production 時,認證走標準 header、授權由 server 從身分推導、發現用公開名片或內部 registry——每一塊都對得上你已經會的後端慣例,沒有新魔法。難的從來不是協議,而是:你那張名片上,值得被別的 AI 呼叫的技能,到底是什麼?