1. 總覽
AI 代理的自主運作、學習和與環境互動能力，可協助達成目標，因此迅速受到歡迎，並徹底改變工作自動化和決策方式。
但具體來說，要如何建構代理程式？本程式碼研究室將協助您入門，說明如何建構可轉換不同國家/地區貨幣的貨幣代理程式。目標是帶您瞭解最新技術，協助您解讀網路上常見的縮寫字 (MCP、ADK、A2A)。
Model Context Protocol (MCP)
Model Context Protocol (MCP) 是一項開放式通訊協定，可將應用程式為 LLM 提供背景資訊的方式標準化。MCP 提供標準化方式，將 AI 模型連結至資源、提示和工具。
Agent Development Kit (ADK)
Agent Development Kit (ADK) 是一個彈性十足的調度框架，可用於開發及部署 AI 代理。ADK 與模型和部署作業無關，且可與其他架構相容。ADK 的設計宗旨是讓代理程式開發更貼近軟體開發，方便開發人員建立、部署及自動調度代理程式架構，處理簡單工作到複雜工作流程等各種任務。
Agent2Agent (A2A) 通訊協定
Agent2Agent (A2A) 通訊協定是一項開放標準，旨在讓 AI 代理順暢通訊及協作。就像 MCP 提供標準方式，讓 LLM 存取資料和工具一樣，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="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.a2a.utils.agent_to_a2a import to_a2a
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."
)
logger.info("--- 🔧 Loading MCP tools from MCP Server... ---")
logger.info("--- 🤖 Creating ADK Currency Agent... ---")
root_agent = 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")
)
)
],
)
如要快速測試貨幣代理程式，可以執行 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],
)
現在就來整合貨幣代理程式，展現 A2A 的強大功能！
8. 公開 Currency Agent A2A 伺服器
ADK 可簡化使用 A2A 通訊協定建構及連結代理的程序。如要將現有的 ADK 代理程式設為可存取 (公開) 的 A2A 伺服器，請使用 ADK 的 to_a2a(root_agent) 函式 (詳情請參閱 ADK 說明文件)。
to_a2a 函式會轉換現有代理程式，使其能與 A2A 搭配運作，並透過 uvicorn 將其公開為伺服器。也就是說，如果您打算將代理程式投入正式環境，就能更嚴格地控管要公開的內容。to_a2a() 函式會使用 A2A Python SDK，根據代理程式碼自動生成代理程式卡。
查看 currency_agent/agent.py 檔案內容，您會發現 to_a2a 的使用方式，以及貨幣代理程式如何以 A2A 伺服器形式公開，而且只需要兩行程式碼！
from google.adk.a2a.utils.agent_to_a2a import to_a2a
# ... see file for full code
# Make the agent A2A-compatible
a2a_app = to_a2a(root_agent, port=10000)
如要執行 A2A 伺服器，請在新終端機中執行下列指令：
uv run uvicorn currency_agent.agent:a2a_app --host localhost --port 10000
如果伺服器成功啟動，輸出內容會如下所示，表示伺服器正在連接埠 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 通訊協定呼叫該程式！
確認遠端代理程式正在執行
如要再次確認代理程式是否正常運作，請前往 to_a2a() 函式自動產生的貨幣代理程式卡片網址。
在瀏覽器中前往 [http://localhost:10000/.well-known/agent.json]
您應該會看到下列代理程式資訊卡：
{
"capabilities": {
},
"defaultInputModes": [
"text/plain"
],
"defaultOutputModes": [
"text/plain"
],
"description": "An agent that can help with currency conversions",
"name": "currency_agent",
"preferredTransport": "JSONRPC",
"protocolVersion": "0.3.0",
"skills": [
{
"description": "An agent that can help with currency conversions I am a specialized assistant for currency conversions. my 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 I 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.",
"id": "currency_agent",
"name": "model",
"tags": [
"llm"
]
},
{
"description": "Use this to get current exchange rate.\n\nArgs:\n currency_from: The currency to convert from (e.g., \"USD\").\n currency_to: The currency to convert to (e.g., \"EUR\").\n currency_date: The date for the exchange rate or \"latest\". Defaults to \"latest\".\n\nReturns:\n A dictionary containing the exchange rate data, or an error message if the request fails.",
"id": "currency_agent-get_exchange_rate",
"name": "get_exchange_rate",
"tags": [
"llm",
"tools"
]
}
],
"supportsAuthenticatedExtendedCard": false,
"url": "http://localhost:10000",
"version": "0.0.1"
}
測試 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:
# Create a resolver to fetch the agent card
resolver = A2ACardResolver(
httpx_client=httpx_client,
base_url=AGENT_URL,
)
agent_card = await resolver.get_agent_card()
# Create a client to interact with the agent
client = A2AClient(
httpx_client=httpx_client,
agent_card=agent_card,
)
print('--- ✅ Connection successful. ---')
await run_single_turn_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"}}}
// ...
--- 🚀 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 帳戶收取本實驗室所用資源的費用，請按照下列步驟操作：