Agent Registry 中的 Agent2Agent (A2A) 代理會宣傳通訊協定介面,其中包含端點網址和通訊協定繫結 (例如 HTTP_JSON)。您可以在註冊資料庫中找到代理的網址,以便從自訂自動化調度管理工具或用戶端呼叫其 A2A 方法。
概覽
| 規格 | 詳細資料 |
|---|---|
| Discovery API | agentregistry.googleapis.com (v1) |
| 叫用 Proxy 主機 | LOCATION-discoveryengine.googleapis.com |
| 通訊協定繫結 | HTTP_JSON |
| 網址專案 ID | Google Cloud 專案編號 (而非專案 ID) |
| 支援的 A2A 方法 | GET /v1/card、POST /v1/message:send、POST /v1/message:stream |
| 必要的訊息結構定義 | message.role = "ROLE_USER",content[].text,不重複messageId |
| 必要 IAM 權限 | roles/agentregistry.viewer (探索) 和 discoveryengine.assistants.assist (叫用) |
事前準備
- 在 Google Cloud 專案中啟用 Agent Registry API (
agentregistry.googleapis.com) 和 Discovery Engine API (discoveryengine.googleapis.com)。 - 如果代理不是直接在 Gemini Enterprise 應用程式中建立,請從 Agent Registry 匯入代理,並授予使用者存取權。如需操作說明,請參閱「從 Agent Registry 匯入 A2A 代理程式」。
- 將適當的身分與存取權管理權限授予呼叫端主體:
- 讀取登錄檔:Agent Registry 檢視者 (
roles/agentregistry.viewer)。 - 如要叫用代理程式:Discovery Engine 編輯者 (
roles/discoveryengine.editor) 或包含discoveryengine.assistants.assist的自訂角色。
- 讀取登錄檔:Agent Registry 檢視者 (
- 如果您使用應用程式預設憑證 (ADC) 進行驗證,請將用戶端設為傳送配額專案標頭:
-H "X-Goog-User-Project: PROJECT_ID"。 - 如要將遠端代理包裝為程式輔助子代理,可以選擇安裝 Agent Development Kit (ADK) 程式庫:
pip install "google-adk[a2a]>=1.29.0"。
步驟 1:探索代理及其 A2A 端點
如要叫用 A2A 代理,請先在代理登錄檔中找出其公告的 url。列出登錄檔位置中的代理 (例如 us 或 eu,以路徑參數形式傳遞至全域 agentregistry.googleapis.com 主機):
curl -X GET \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
"https://agentregistry.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/agents?pageSize=100"
您也可以使用 Google Cloud CLI,依顯示名稱前置字串搜尋代理:
gcloud agent-registry agents search \
--project=PROJECT_ID \
--location=LOCATION \
--search-string="displayName:My_Agent_*"
在傳回的代理資源中,檢查 protocols 陣列。找出 type 等於 A2A_AGENT 且 interfaces[].protocolBinding 等於 HTTP_JSON 的項目。擷取對應的 url:
{
"name": "projects/PROJECT_ID/locations/LOCATION/agents/AGENT_RESOURCE_ID",
"displayName": "My Agent",
"protocols": [
{
"type": "A2A_AGENT",
"protocolVersion": "0.3.0",
"interfaces": [
{
"url": "https://LOCATION-discoveryengine.googleapis.com/v1/projects/PROJECT_NUMBER/locations/LOCATION/collections/default_collection/engines/ENGINE_ID/assistants/default_assistant/agents/AGENT_ID/a2a",
"protocolBinding": "HTTP_JSON"
}
]
}
]
}
如需完整的代理資源結構定義,請參閱 projects.locations.agents REST API 參考資料。
步驟 2:擷取代理程式卡片
代理資訊卡會提供中繼資料,說明代理的身分、說明和輸入/輸出功能。如要擷取卡片,請將 HTTP GET 要求傳送至 /v1/card 路徑,並附加至代理商的 A2A 端點網址:
curl -X GET \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
"A2A_ENDPOINT_URL/v1/card"
回應酬載範例:
{
"name": "My Agent",
"description": "What the agent does.",
"url": "A2A_ENDPOINT_URL",
"capabilities": {},
"defaultInputModes": ["text"],
"defaultOutputModes": ["text"],
"preferredTransport": "HTTP+JSON"
}
步驟 3:傳送訊息
如要將使用者查詢傳送至代理,請向 /v1/message:send 發出 POST 要求。要求主體必須符合 A2A 訊息結構定義,其中 role 須設為 ROLE_USER、content 陣列須包含文字部分,且 messageId 須為不重複產生:
curl -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
"A2A_ENDPOINT_URL/v1/message:send" \
-d '{
"message": {
"role": "ROLE_USER",
"content": [
{
"text": "What can you help me with?"
}
],
"messageId": "UNIQUE_UUID_STRING"
}
}'
在回應酬載中,服務專員的回覆會傳回 message 物件:
{
"message": {
"contextId": "projects/PROJECT_NUMBER/locations/LOCATION/collections/default_collection/engines/ENGINE_ID/sessions/SESSION_ID",
"role": "ROLE_AGENT",
"content": [
{
"text": "I am an AI assistant..."
}
]
}
}
串連 content[].text 中的文字字串,顯示完整回覆。如要在同一個工作階段中繼續對話,請儲存傳回的 contextId 字串,並在下一個要求中以 message.contextId 的形式提供。
如需完整的訊息酬載結構定義,請參閱 A2A message:send REST API 參考資料。
逐句回覆
如要串流輸出內容,請傳送 POST 要求,並附上與 /v1/message:stream 相同的訊息主體:
curl -N -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json" \
"A2A_ENDPOINT_URL/v1/message:stream" \
-d '{
"message": {
"role": "ROLE_USER",
"content": [
{
"text": "Say hello."
}
],
"messageId": "UNIQUE_UUID_STRING"
}
}'
端點會透過 HTTP 傳回串流區塊物件的 JSON 陣列。在 content[].text 片段抵達時依序附加。串流區塊也包含 metadata.sessionInfo 和 metadata.assistToken。
如需串流酬載規格,請參閱 A2A message:stream REST API 參考資料。
使用 Python 呼叫 A2A 端點
這個 Python 指令碼會解析 Agent Registry 中的 A2A 端點,並使用原始 HTTP 要求傳送訊息:
# Install dependencies: pip install google-auth requests
import uuid
import google.auth
from google.auth.transport.requests import AuthorizedSession
# TODO(developer): Replace placeholder values with your project ID and location.
project_id = "PROJECT_ID"
location = "LOCATION" # Registry location (for example: "us" or "eu")
target_display_name = "My Agent"
query_text = "What can you help me with?"
# Initialize credentials and authorized session
creds, _ = google.auth.default(
scopes=["https://www.googleapis.com/auth/cloud-platform"]
)
session = AuthorizedSession(creds)
# Step 1: Resolve the A2A endpoint URL from the Agent Registry
registry_url = (
f"https://agentregistry.googleapis.com/v1/"
f"projects/{project_id}/locations/{location}/agents"
)
response = session.get(registry_url)
response.raise_for_status()
agents = response.json().get("agents", [])
def get_a2a_url(agent_resource):
for proto in agent_resource.get("protocols") or []:
if proto.get("type") == "A2A_AGENT":
for iface in proto.get("interfaces", []):
if iface.get("protocolBinding") == "HTTP_JSON":
return iface.get("url")
return None
target_agent = next(
(a for a in agents if a.get("displayName") == target_display_name),
None
)
if not target_agent:
raise SystemExit(f"Agent '{target_display_name}' not found in registry.")
endpoint_url = get_a2a_url(target_agent)
if not endpoint_url:
raise SystemExit("Target agent does not publish an HTTP_JSON A2A endpoint.")
# Step 2: Fetch and verify the agent card
card_resp = session.get(f"{endpoint_url}/v1/card")
card_resp.raise_for_status()
card = card_resp.json()
print("Resolved Agent:", card.get("name"))
# Step 3: Send an A2A message
body = {
"message": {
"role": "ROLE_USER",
"content": [{"text": query_text}],
"messageId": str(uuid.uuid4()),
}
}
send_resp = session.post(f"{endpoint_url}/v1/message:send", json=body)
send_resp.raise_for_status()
reply_message = send_resp.json().get("message", {})
full_reply_text = "".join(
part.get("text", "") for part in reply_message.get("content", [])
)
print("Agent Reply:", full_reply_text)
使用 ADK 簡化自動化調度管理
Agent Development Kit (ADK) 會自動解析登錄端點,並將遠端 A2A 代理程式包裝為子代理程式:
from google.adk.integrations.agent_registry import AgentRegistry
# Initialize registry client
registry = AgentRegistry(project_id="PROJECT_ID", location="LOCATION")
# Resolve remote A2A agent directly by resource name
remote_agent = registry.get_remote_a2a_agent(
agent_name="agents/AGENT_RESOURCE_ID"
)
其他注意事項
A2A 端點的行為如下:
- 嚴格的路徑命名:HTTP+JSON 繫結僅支援
GET {url}/v1/card、POST {url}/v1/message:send和POST {url}/v1/message:stream。 - 嚴格結構定義驗證:如果將純
"user"做為角色傳遞,會傳回 HTTP400 Bad Request錯誤。您必須傳遞列舉字串"ROLE_USER"。同樣地,訊息文字必須位於content陣列內,而非parts,且嚴格來說,messageId為必要項目。 - 非 A2A 代理叫用錯誤:如果代理在註冊表中缺少
A2A_AGENT通訊協定項目 (例如某些預先建構或代管的代理),對其 Proxy 網址呼叫getCard會傳回501 UNIMPLEMENTED(「... is not supported yet」),呼叫message:send則會傳回400 INVALID_ARGUMENT(「Unsupported agent」)。 - 網址中的專案編號:登錄檔會傳回包含專案編號 (而非專案 ID) 的 A2A 網址。提出 HTTP 要求時,請勿變更這個數字字串。
疑難排解
請參閱下表,瞭解如何排解常見的 A2A 端點錯誤:
| 問題 | 可能原因 | 解析度 |
|---|---|---|
getCard 要求傳回 HTTP 404 |
使用不正確的路徑別名 (例如 /v1:getCard 或 /.well-known/agent-card.json)。 |
請嚴格將 GET 要求傳送至 GET {url}/v1/card。 |
| HTTP 400 *「Unknown name 'parts'」(名稱「parts」不明)* | 使用舊版或生成式 AI 用戶端內文格式。 | 請將文字字串放在 content 內,而非 parts。 |
HTTP 400 role 的列舉值無效 |
傳遞小寫 "user" 或 "user_role"。 |
將 message.role 設為 "ROLE_USER"。 |
| HTTP 501「尚未支援」 | 在未發布 A2A 介面的代理上呼叫 getCard。 |
呼叫前,請檢查登錄檔資源的 protocols 陣列,確認是否支援 A2A_AGENT。 |
| HTTP 400 *「不支援的代理程式」* | 在非 A2A 代理上呼叫 message:send。 |
選擇登錄檔定義包含有效 A2A_AGENT 通訊協定繫結的代理。 |
HTTP 401或HTTP 403 Permission Denied |
缺少 OAuth 範圍、IAM 角色或配額專案標頭。 | 檢查呼叫端 IAM 角色 (agentregistry.viewer 和 assistants.assist);驗證 cloud-platform 範圍;如果使用 ADC,請傳遞 -H "X-Goog-User-Project: PROJECT_ID"。 |