開始使用 MCP、ADK 和 A2A

開始使用 MCP、ADK 和 A2A

程式碼研究室簡介

subject上次更新時間:6月 25, 2025
account_circle作者:Jack Wotherspoon

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 伺服器

軟硬體需求

  • 瀏覽器,例如 ChromeFirefox
  • 啟用計費功能的 Google Cloud 專案。

2. 事前準備

建立專案

  1. Google Cloud 控制台的專案選取器頁面中,選取或建立 Google Cloud 專案
  2. 確認 Cloud 專案已啟用計費功能。瞭解如何檢查專案是否已啟用計費功能
  3. 按一下這個連結,啟動 Cloud Shell。如要在 Cloud Shell 終端機 (用於執行雲端指令) 和編輯器 (用於建構專案) 之間切換,請點選 Cloud Shell 中的對應按鈕。
  4. 連線至 Cloud Shell 後,請使用下列指令檢查您是否已通過驗證,且專案已設為您的專案 ID:
gcloud auth list
  1. 在 Cloud Shell 中執行下列指令,確認 gcloud 指令已瞭解您的專案。
gcloud config list project
  1. 使用下列指令設定專案:
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
  1. 使用下列指令啟用必要的 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
  1. 請務必使用 Python 3.10 以上版本

如要瞭解 gcloud 指令和用法,請參閱說明文件

3. 安裝

  1. 複製存放區:
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
  1. 安裝 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"
  1. 設定環境變數 (透過 .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 做為代理程式。

ADK 網頁版 UI

在對話區域中向代理程式提問,例如「250 加幣等於多少美元?」。在提供回覆前,您應該會看到代理程式呼叫我們的 get_exchange_rate MCP 工具。

ADK Web Currency Agent

代理程式運作正常!可處理與貨幣換算 💸 相關的查詢。

7. Agent2Agent (A2A) 通訊協定

Agent2Agent (A2A) 通訊協定是開放式標準,旨在讓 AI 代理程式之間能夠順暢通訊及協作。這樣一來,使用不同架構和由不同供應商建構的代理,就能以通用語言互相溝通,打破資料孤島並促進互通性。

A2A 通訊協定

A2A 可讓代理程式執行下列操作:

  • 探索:使用標準化的代理程式資訊卡,尋找其他代理程式並瞭解其技能 (AgentSkill) 和功能 (AgentCapabilities)。
  • 通訊:安全地交換訊息和資料。
  • 協作:指派工作及協調行動,以達成複雜目標。

A2A 通訊協定透過「代理程式資訊卡」等機制促進這類通訊,代理程式可使用這類資訊卡做為數位名片,宣傳自身功能和連線資訊。

A2A 代理資訊卡

現在請使用 A2A 公開幣別代理程式,以便其他代理程式和用戶端呼叫。

A2A Python SDK

A2A Python SDK 為上述各項資源提供 Pydantic 模型:AgentSkillAgentCapabilitiesAgentCard。這個介面可加快開發速度,並與 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 帳戶收取本實驗室所用資源的費用,請按照下列步驟操作:

  1. 在 Google Cloud 控制台中前往「管理資源」頁面。
  2. 在專案清單中選取要刪除的專案,然後點按「刪除」。
  3. 在對話方塊中輸入專案 ID,然後按一下「Shut down」(關閉) 即可刪除專案。