程式碼研究室簡介
1. 總覽
AI 代理的自主運作、學習和與環境互動能力,可協助達成目標,因此迅速受到歡迎,並徹底改變工作自動化和決策方式。
但具體來說,要如何建構代理程式?本程式碼研究室將協助您入門,說明如何建構可轉換不同國家/地區貨幣的代理程式。目標是帶您瞭解最新技術,協助您解讀網路上常見的縮寫字 (MCP、ADK、A2A)。
Model Context Protocol (MCP)
模型脈絡通訊協定 (MCP) 是一種開放式通訊協定,可標準化應用程式提供 LLM 脈絡的方式。MCP 提供標準化方式,可將 AI 模型連結至資源、提示和工具。
Agent Development Kit (ADK)
Agent Development Kit (ADK) 是一個彈性十足的協調架構,可用於開發及部署 AI 代理。ADK 與模型和部署作業無關,且可與其他架構相容。ADK 的設計宗旨是讓代理程式開發更貼近軟體開發,方便開發人員建立、部署及協調代理程式架構,處理簡單工作到複雜工作流程等各種任務。
Agent2Agent (A2A) 通訊協定
Agent2Agent (A2A) 通訊協定是開放式標準,旨在讓 AI 代理程式之間能夠順暢通訊及協作。就像 MCP 提供標準化方式,讓大型語言模型存取資料和工具一樣,A2A 也提供標準化方式,讓代理程式與其他代理程式對話!在代理程式由不同供應商使用各種架構建構的世界中,A2A 提供通用語言,打破資料孤島並促進互通性。
課程內容
- 如何建立本機 MCP 伺服器
- 將 MCP 伺服器部署至 Cloud Run
- 如何使用 Agent Development Kit 建構代理,並使用 MCP 工具
- 如何將 ADK 代理程式公開為 A2A 伺服器
- 使用 A2A 用戶端測試 A2A 伺服器
軟硬體需求
2. 事前準備
建立專案
- 在 Google Cloud 控制台的專案選取器頁面中,選取或建立 Google Cloud 專案。
- 確認 Cloud 專案已啟用計費功能。瞭解如何檢查專案是否已啟用計費功能。
- 按一下這個連結,啟動 Cloud Shell。如要在 Cloud Shell 終端機 (用於執行雲端指令) 和編輯器 (用於建構專案) 之間切換,請點選 Cloud Shell 中的對應按鈕。
- 連線至 Cloud Shell 後,請使用下列指令檢查您是否已通過驗證,且專案已設為您的專案 ID:
gcloud auth list
- 在 Cloud Shell 中執行下列指令,確認 gcloud 指令已瞭解您的專案。
gcloud config list project
- 使用下列指令設定專案:
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- 使用下列指令啟用必要的 API。這項作業可能需要幾分鐘才能完成。
gcloud services enable cloudresourcemanager.googleapis.com \
servicenetworking.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
- 請務必使用 Python 3.10 以上版本
如要瞭解 gcloud 指令和用法,請參閱說明文件。
3. 安裝
- 複製存放區:
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
- 安裝 uv (用於管理依附元件):
# macOS and Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows (uncomment below line)
# powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
- 設定環境變數 (透過
.env
檔案):
執行下列指令,建立 .env
檔案:
echo "GOOGLE_GENAI_USE_VERTEXAI=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env
4. 建立本機 MCP 伺服器
在協調貨幣代理程式之前,您必須先建立 MCP 伺服器,以公開代理程式所需的工具。
MCP 伺服器可讓您編寫輕量型程式,將特定功能 (例如擷取匯率) 顯示為工具。代理程式 (甚至多個代理程式) 接著就能使用標準化的 Model Context Protocol (MCP) 存取這些工具。
您可以運用 FastMCP Python 套件建立 MCP 伺服器,公開名為 get_exchange_rate
的單一工具。get_exchange_rate
工具會透過網際網路呼叫 Frankfurter API,取得兩種貨幣之間的目前匯率。
MCP 伺服器的程式碼位於 mcp-server/server.py
檔案中:
import logging
import os
import httpx
from fastmcp import FastMCP
# Set up logging
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
mcp = FastMCP("Currency MCP Server 💵")
@mcp.tool()
def get_exchange_rate(
currency_from: str = 'USD',
currency_to: str = 'EUR',
currency_date: str = 'latest',
):
"""Use this to get current exchange rate.
Args:
currency_from: The currency to convert from (e.g., "USD").
currency_to: The currency to convert to (e.g., "EUR").
currency_date: The date for the exchange rate or "latest". Defaults to "latest".
Returns:
A dictionary containing the exchange rate data, or an error message if the request fails.
"""
logger.info(f"--- 🛠️ Tool: get_exchange_rate called for converting {currency_from} to {currency_to} ---")
try:
response = httpx.get(
f'https://api.frankfurter.app/{currency_date}',
params={'from': currency_from, 'to': currency_to},
)
response.raise_for_status()
data = response.json()
if 'rates' not in data:
return {'error': 'Invalid API response format.'}
logger.info(f'✅ API response: {data}')
return data
except httpx.HTTPError as e:
return {'error': f'API request failed: {e}'}
except ValueError:
return {'error': 'Invalid JSON response from API.'}
if __name__ == "__main__":
logger.info(f"🚀 MCP server started on port {os.getenv('PORT', 8080)}")
# Could also use 'sse' transport, host="0.0.0.0" required for Cloud Run.
asyncio.run(
mcp.run_async(
transport="streamable-http",
host="0.0.0.0",
port=os.getenv("PORT", 8080),
)
)
如要在本機啟動 MCP 伺服器,請開啟終端機並執行下列指令 (伺服器會在 http://localhost:8080
上啟動):
uv run mcp-server/server.py
測試 MCP 伺服器是否正常運作,以及是否可使用 Model Context Protocol 存取 get_exchange_rate
工具。
在新的終端機視窗中 (以免停止本機 MCP 伺服器),執行下列指令:
uv run mcp-server/test_server.py
您應該會看到目前 1 美元兌換歐元的匯率輸出:
--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
--- ✅ Success: {
"amount": 1.0,
"base": "USD",
"date": "2025-05-26",
"rates": {
"EUR": 0.87866
}
} ---
太棒了!您已成功建立可正常運作的 MCP 伺服器,並提供代理程式可存取的工具。
在前往下一個工作站之前,請在啟動 MCP 伺服器的終端機中執行 Ctrl+C
(或 Mac 上的 Command+C
),停止在本機執行的 MCP 伺服器。
5. 將 MCP 伺服器部署至 Cloud Run
現在您已準備好將 MCP 伺服器部署為遠端 MCP 伺服器至 Cloud Run 🚀☁️
遠端執行 MCP 伺服器的優點
在 Cloud Run 上遠端執行 MCP 伺服器可帶來多項優點:
- 📈擴充性:Cloud Run 的設計宗旨是快速擴充,以便處理所有傳入要求。Cloud Run 會根據需求自動調度 MCP 伺服器資源。
- 👥集中式伺服器:您可以透過 IAM 權限與團隊成員共用集中式 MCP 伺服器的存取權,讓他們從本機連線至該伺服器,而不必在本機執行自己的伺服器。如果 MCP 伺服器有任何變更,所有團隊成員都會受益。
- 🔐安全性:Cloud Run 可輕鬆強制執行已驗證的要求。這樣一來,只有安全連線才能連上 MCP 伺服器,防止未經授權的存取。
變更至 mcp-server
目錄:
cd mcp-server
將 MCP 伺服器部署至 Cloud Run:
gcloud run deploy mcp-server --no-allow-unauthenticated --region=us-central1 --source .
如果服務部署成功,您會看到類似以下的訊息:
Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.
驗證 MCP 用戶端
由於您指定 --no-allow-unauthenticated
需要驗證,因此連線至遠端 MCP 伺服器的任何 MCP 用戶端都必須通過驗證。
如要進一步瞭解這個主題,請參閱「在 Cloud Run 上代管 MCP 伺服器」官方文件,瞭解 MCP 用戶端執行位置的相關資訊。
您必須執行 Cloud Run Proxy,在本機電腦上建立通過驗證的通道,連線至遠端 MCP 伺服器。
根據預設,Cloud Run 服務的網址會要求所有要求都必須透過 Cloud Run Invoker (roles/run.invoker
) IAM 角色授權。這項 IAM 政策繫結可確保使用強大的安全機制驗證本機 MCP 用戶端。
請確認您或任何嘗試存取遠端 MCP 伺服器的團隊成員,都已將 roles/run.invoker
IAM 角色繫結至 IAM 主體 (Google Cloud 帳戶)。
gcloud run services proxy mcp-server --region=us-central1
您應該會看到以下的輸出內容:
Proxying to Cloud Run service [mcp-server] in project [<YOUR_PROJECT_ID>] region [us-central1]
http://127.0.0.1:8080 proxies to https://mcp-server-abcdefgh-uc.a.run.app
傳送至 http://127.0.0.1:8080
的所有流量,現都會經過驗證並轉送至遠端 MCP 伺服器。
測試遠端 MCP 伺服器
在新終端機中返回根資料夾,然後重新執行 mcp-server/test_server.py
檔案,確認遠端 MCP 伺服器是否正常運作。
cd ..
uv run mcp-server/test_server.py
您應該會看到與在本機執行伺服器時類似的輸出內容:
--- 🛠️ Tool found: get_exchange_rate ---
--- 🪛 Calling get_exchange_rate tool for USD to EUR ---
--- ✅ Success: {
"amount": 1.0,
"base": "USD",
"date": "2025-05-26",
"rates": {
"EUR": 0.87866
}
} ---
如要確認是否確實呼叫遠端伺服器,可以查詢已部署的 Cloud Run MCP 伺服器的記錄:
gcloud run services logs read mcp-server --region us-central1 --limit 5
您應該會在記錄中看到下列輸出內容:
2025-06-04 14:28:29,871 [INFO]: --- 🛠️ Tool: get_exchange_rate called for converting USD to EUR ---
2025-06-04 14:28:30,610 [INFO]: HTTP Request: GET https://api.frankfurter.app/latest?from=USD&to=EUR "HTTP/1.1 200 OK"
2025-06-04 14:28:30,611 [INFO]: ✅ API response: {'amount': 1.0, 'base': 'USD', 'date': '2025-06-03', 'rates': {'EUR': 0.87827}}
您現在有了遠端 MCP 伺服器,接下來可以建立代理程式!🤖
6. 使用 Agent Development Kit (ADK) 建立代理
您已部署 MCP 伺服器,現在可以使用 Agent Development Kit (ADK) 建立貨幣代理。
Agent Development Kit 最近發布了 1.0.0 穩定版。這項里程碑代表 Python ADK 現在已可投入生產,為開發人員提供可靠且穩固的平台,讓他們有信心在實際環境中建構及部署代理程式。
ADK 可建立極輕量的代理程式,並內建支援 MCP 工具,方便代理程式連線至 MCP 伺服器。貨幣代理程式會使用 ADK 的 MCPToolset 類別存取 get_exchange_rate
工具。
貨幣代理商的代碼位於 currency_agent/agent.py
:
import logging
import os
from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import MCPToolset, StreamableHTTPConnectionParams
logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)
load_dotenv()
SYSTEM_INSTRUCTION = (
"You are a specialized assistant for currency conversions. "
"Your sole purpose is to use the 'get_exchange_rate' tool to answer questions about currency exchange rates. "
"If the user asks about anything other than currency conversion or exchange rates, "
"politely state that you cannot help with that topic and can only assist with currency-related queries. "
"Do not attempt to answer unrelated questions or use tools for other purposes."
)
def create_agent() -> LlmAgent:
"""Constructs the ADK currency conversion agent."""
logger.info("--- 🔧 Loading MCP tools from MCP Server... ---")
logger.info("--- 🤖 Creating ADK Currency Agent... ---")
return LlmAgent(
model="gemini-2.5-flash",
name="currency_agent",
description="An agent that can help with currency conversions",
instruction=SYSTEM_INSTRUCTION,
tools=[
MCPToolset(
connection_params=StreamableHTTPConnectionParams(
url=os.getenv("MCP_SERVER_URL", "http://localhost:8080/mcp")
)
)
],
)
root_agent = create_agent()
如要快速測試貨幣代理程式,可以執行 adk web
,透過 ADK 的開發人員 UI 存取:
uv run adk web
在瀏覽器中前往 http://localhost:8000
,即可查看及測試代理程式!
確認網頁使用者介面左上角已選取 currency_agent
做為代理程式。
在對話區域中向代理程式提問,例如「250 加幣等於多少美元?」。在提供回覆前,您應該會看到代理程式呼叫我們的 get_exchange_rate
MCP 工具。
代理程式運作正常!可處理與貨幣換算 💸 相關的查詢。
7. Agent2Agent (A2A) 通訊協定
Agent2Agent (A2A) 通訊協定是開放式標準,旨在讓 AI 代理程式之間能夠順暢通訊及協作。這樣一來,使用不同架構和由不同供應商建構的代理,就能以通用語言互相溝通,打破資料孤島並促進互通性。
A2A 可讓代理程式執行下列操作:
- 探索:使用標準化的代理程式資訊卡,尋找其他代理程式並瞭解其技能 (AgentSkill) 和功能 (AgentCapabilities)。
- 通訊:安全地交換訊息和資料。
- 協作:指派工作及協調行動,以達成複雜目標。
A2A 通訊協定透過「代理程式資訊卡」等機制促進這類通訊,代理程式可使用這類資訊卡做為數位名片,宣傳自身功能和連線資訊。
現在請使用 A2A 公開幣別代理程式,以便其他代理程式和用戶端呼叫。
A2A Python SDK
A2A Python SDK 為上述各項資源提供 Pydantic 模型:AgentSkill、AgentCapabilities 和 AgentCard。這個介面可加快開發速度,並與 A2A 通訊協定整合。
AgentSkill
:向其他代理商宣傳貨幣代理商有工具可供使用的方式 get_exchange_rate
:
# A2A Agent Skill definition
skill = AgentSkill(
id='get_exchange_rate',
name='Currency Exchange Rates Tool',
description='Helps with exchange values between various currencies',
tags=['currency conversion', 'currency exchange'],
examples=['What is exchange rate between USD and GBP?'],
)
然後,在 AgentCard
中,系統會列出代理程式的技能和功能,以及代理程式可處理的輸入和輸出模式等其他詳細資料:
# A2A Agent Card definition
agent_card = AgentCard(
name='Currency Agent',
description='Helps with exchange rates for currencies',
url=f'http://{host}:{port}/',
version='1.0.0',
defaultInputModes=["text"],
defaultOutputModes=["text"],
capabilities=AgentCapabilities(streaming=True),
skills=[skill],
)
AgentExecutor 介面會處理 A2A 代理處理要求及產生回應/事件的核心邏輯。A2A Python SDK 提供抽象基本類別 a2a.server.agent_execution.AgentExecutor
,您必須實作這個類別。
現在是時候整合所有內容,並透過貨幣代理展示 A2A 的強大功能!
8. Currency Agent A2A Server
現在,您將查看一些程式碼片段,瞭解構成 A2A 伺服器的不同部分如何組合在一起。
查看 currency_agent/agent_executor.py
檔案內部,您會發現繼承自 A2A 抽象 AgentExecutor
類別的 ADKAgentExecutor
類別。這個類別會透過叫用 ADK 執行器來處理 ADK 代理的呼叫作業、處理對代理的要求,以及在 ADK 使用的 google.genai.types
和 A2A 使用的 a2a.types
之間來回轉換。
# ... see file for full code
class ADKAgentExecutor(AgentExecutor):
"""An AgentExecutor that runs an ADK agent."""
def __init__(self, runner: Runner, card: AgentCard):
self.runner = runner
self._card = card
self._running_sessions = {}
def _run_agent(
self, session_id, new_message: types.Content
) -> AsyncGenerator[Event, None]:
return self.runner.run_async(
session_id=session_id, user_id="self", new_message=new_message
)
async def _process_request(
self,
new_message: types.Content,
session_id: str,
task_updater: TaskUpdater,
) -> None:
session = await self._upsert_session(
session_id,
)
session_id = session.id
# Run through all events within the request.
async for event in self._run_agent(session_id, new_message):
if event.is_final_response():
parts = convert_genai_parts_to_a2a(event.content.parts)
logger.debug("✅ Yielding final response: %s", parts)
await task_updater.add_artifact(parts)
await task_updater.complete()
break
# If the agent is not making a function call, yield an update.
if not event.get_function_calls():
logger.debug("⏳ Yielding update response")
await task_updater.update_status(
TaskState.working,
message=task_updater.new_agent_message(
convert_genai_parts_to_a2a(event.content.parts),
),
)
else:
logger.debug("➡️ Skipping event")
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
):
# Run the agent until either complete or the task is suspended.
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
# Immediately notify that the task is submitted.
if not context.current_task:
updater.submit()
updater.start_work()
await self._process_request(
types.UserContent(
parts=convert_a2a_parts_to_genai(context.message.parts),
),
context.context_id,
updater,
)
logger.debug("--- 💵💱💶 [Currency] execute exiting ---")
# ... see file for full code
在 currency_agent/__main__.py
內,您可以初始化 AgentSkill、AgentCard,並建立 ADK 貨幣代理程式。您也可以在這裡設定及啟動 A2A 伺服器。
A2A Python SDK 提供 A2AFastAPIApplication
類別,可簡化執行符合 A2A 規範的 HTTP 伺服器。這個應用程式使用 FastAPI 做為網路架構,通常會搭配 ASGI 伺服器 (例如 Uvicorn) 執行。
# ... see file for full code
@click.command()
@click.option("--host", "host", default="localhost")
@click.option("--port", "port", default=10000)
def main(host: str, port: int):
# Verify one of Google AI Studio or Vertex AI is being used
if os.getenv("GOOGLE_GENAI_USE_VERTEXAI") != "TRUE" and not os.getenv(
"GOOGLE_API_KEY"
):
raise ValueError(
"GOOGLE_API_KEY environment variable not set and "
"GOOGLE_GENAI_USE_VERTEXAI is not TRUE."
)
# A2A Agent Skill definition
skill = AgentSkill(
id="get_exchange_rate",
name="Currency Exchange Rates Tool",
description="Helps with exchange values between various currencies",
tags=["currency conversion", "currency exchange"],
examples=["What is exchange rate between USD and GBP?"],
)
# A2A Agent Card definition
agent_card = AgentCard(
name="Currency Agent",
description="Helps with exchange rates for currencies",
url=f"http://{host}:{port}/",
version="1.0.0",
defaultInputModes=["text"],
defaultOutputModes=["text"],
capabilities=AgentCapabilities(streaming=True),
skills=[skill],
)
# Create the ADK runner and executor.
runner = Runner(
app_name=agent_card.name,
agent=root_agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
agent_executor = ADKAgentExecutor(runner, agent_card)
request_handler = DefaultRequestHandler(
agent_executor=agent_executor,
task_store=InMemoryTaskStore(),
)
server = A2AFastAPIApplication(
agent_card=agent_card, http_handler=request_handler
)
uvicorn.run(server.build(), host=host, port=port)
# ... see file for full code
如要執行 A2A 伺服器,請在新終端機中執行下列指令:
uv run currency_agent/
如果伺服器成功啟動,輸出內容會如下所示,表示伺服器正在連接埠 10000 上執行:
[INFO]: --- 🔧 Loading MCP tools from MCP Server... ---
[INFO]: --- 🤖 Creating ADK Currency Agent... ---
INFO: Started server process [45824]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://localhost:10000 (Press CTRL+C to quit)
貨幣代理程式現在已成功以 A2A 伺服器身分執行,其他代理程式或用戶端可以使用 A2A 通訊協定呼叫該程式!
測試 A2A 伺服器
現在可以使用 A2A 向伺服器傳送一些要求,測試伺服器!
A2A Python SDK 提供 a2a.client.A2AClient
類別,可簡化這項程序。
檔案 currency_agent/test_client.py
包含的程式碼會針對 A2A 伺服器執行多個不同的測試案例。
# ... see file for full code
# Example test using A2AClient
async def run_single_turn_test(client: A2AClient) -> None:
"""Runs a single-turn non-streaming test."""
send_message_payload = create_send_message_payload(text="how much is 100 USD in CAD?")
request = SendMessageRequest(
id=str(uuid4()), params=MessageSendParams(**send_message_payload)
)
print("--- ✉️ Single Turn Request ---")
# Send Message
response: SendMessageResponse = await client.send_message(request)
print_json_response(response, "📥 Single Turn Request Response")
if not isinstance(response.root, SendMessageSuccessResponse):
print("received non-success response. Aborting get task ")
return
if not isinstance(response.root.result, Task):
print("received non-task response. Aborting get task ")
return
task_id: str = response.root.result.id
print("--- ❔ Query Task ---")
# query the task
get_request = GetTaskRequest(id=str(uuid4()), params=TaskQueryParams(id=task_id))
get_response: GetTaskResponse = await client.get_task(get_request)
print_json_response(get_response, "📥 Query Task Response")
# ----- Main Entrypoint (Create client --> Run tests) -----
async def main() -> None:
"""Main function to run the tests."""
print(f'--- 🔄 Connecting to agent at {AGENT_URL}... ---')
try:
async with httpx.AsyncClient() as httpx_client:
client = await A2AClient.get_client_from_agent_card_url(
httpx_client, AGENT_URL
)
print('--- ✅ Connection successful. ---')
await run_single_turn_test(client)
await run_streaming_test(client)
await run_multi_turn_test(client)
except Exception as e:
traceback.print_exc()
print(f'--- ❌ An error occurred: {e} ---')
print('Ensure the agent server is running.')
使用下列指令執行測試:
uv run currency_agent/test_client.py
測試執行成功會產生下列結果:
--- 🔄 Connecting to agent at http://localhost:10000... ---
--- ✅ Connection successful. ---
--- ✉️ Single Turn Request ---
--- 📥 Single Turn Request Response ---
{"id":"3bc92d7b-d857-4e93-9ff0-b2fb865f6e35","jsonrpc":"2.0","result":{"artifacts":[{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"Based on the current exchange rate, 1 USD is equivalent to 1.3704 CAD. Therefore, 100 USD would be 137.04 CAD.\n"}]}],"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","history":[{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"59819269f7d04849b0bfca7d43ec073c","parts":[{"kind":"text","text":"how much is 100 USD in CAD?"}],"role":"user","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"},{"contextId":"2d66f277-152c-46ef-881d-7fe32866e9f5","kind":"message","messageId":"286095c6-12c9-40cb-9596-a9676d570dbd","parts":[],"role":"agent","taskId":"52ae2392-84f5-429a-a14b-8413d3d20d97"}],"id":"52ae2392-84f5-429a-a14b-8413d3d20d97","kind":"task","status":{"state":"completed"}}}
// ...
--- ⏩ Single Turn Streaming Request ---
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"submitted"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","final":false,"kind":"status-update","status":{"message":{"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"message","messageId":"25f5f972-9475-4e4a-a08d-e13f521d7462","parts":[],"role":"agent","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"},"state":"working"},"taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
--- ⏳ Streaming Chunk ---
{"id":"21239a5f-abbf-4a5e-a249-c101eb1dfbdd","jsonrpc":"2.0","result":{"artifact":{"artifactId":"35e89e14-b977-4397-a23b-92c84bc32379","parts":[{"kind":"text","text":"The current exchange rate is 1 EUR to 164.15 JPY. So, 50 EUR would be 8207.5 JPY.\n"}]},"contextId":"f268ad8c-b3bf-4439-9a64-5e02dfbb9a62","kind":"artifact-update","taskId":"761d2275-d58b-46f8-9c8d-68cd72e0667d"}}
// ...
--- 🚀 First turn completed, no further input required for this test case. ---
成功了!您已成功測試,可透過 A2A 伺服器與貨幣代理程式通訊!🎉
請前往 GitHub 查看 a2a-samples 存放區,瞭解更多進階用途!
想部署代理程式嗎?Vertex AI Agent Engine 提供代管服務,可將 AI 代理部署至正式環境!
9. 恭喜
恭喜!您已成功建構及部署遠端 MCP 伺服器,並使用 Agent Development Kit (ADK) 建立可透過 MCP 連線至工具的貨幣代理,以及使用 Agent2Agent (A2A) 通訊協定公開代理!現在可以使用 A2A,讓貨幣代理與任何架構的其他代理互動!
如要查看完整程式碼說明文件,請按這裡。
涵蓋內容
- 如何建立本機 MCP 伺服器
- 將 MCP 伺服器部署至 Cloud Run
- 如何使用 Agent Development Kit 建構代理,並使用 MCP 工具
- 如何將 ADK 代理程式公開為 A2A 伺服器
- 使用 A2A 用戶端測試 A2A 伺服器
清除所用資源
如要避免系統向您的 Google Cloud 帳戶收取本實驗室所用資源的費用,請按照下列步驟操作: