关于此 Codelab
1. 概览
AI 智能体正迅速受到欢迎，它们能够自主运行、学习并与环境互动以实现目标，从而彻底改变任务自动化和决策制定方式。
但具体来说，如何构建代理呢？此 Codelab 将向您展示如何构建一个可以进行不同国家/地区货币之间转换的货币代理，帮助您入门。本文旨在介绍最新技术，帮助您了解可能在网上看到的一些缩写词（MCP、ADK、A2A）。
模型上下文协议 (MCP)
模型上下文协议 (MCP) 是一种开放协议，可标准化应用向 LLM 提供上下文的方式。MCP 提供了一种将 AI 模型连接到资源、提示和工具的标准化方式。
Agent Development Kit (ADK)
智能体开发套件 (ADK) 是一个灵活的编排框架，用于开发和部署 AI 智能体。ADK 与模型无关、与部署无关，并且旨在与其他框架兼容。ADK 的设计旨在让智能体开发更像软件开发，从而让开发者更轻松地创建、部署和编排从简单任务到复杂工作流的智能体架构。
Agent2Agent (A2A) 协议
Agent2Agent (A2A) 协议是一种开放标准，旨在实现 AI 代理之间的无缝通信和协作。正如 MCP 提供了一种标准化方式，让 LLM 能够访问数据和工具一样，A2A 也提供了一种标准化方式，让代理能够与其他代理对话！在一个代理由不同供应商使用各种框架构建的世界中，A2A 提供了一种通用语言，打破了孤岛并促进了互操作性。
学习内容
- 如何创建本地 MCP 服务器
- 将 MCP 服务器部署到 Cloud Run
- 如何使用智能体开发套件构建使用 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 服务器，您可以编写轻量级程序，将特定功能（例如获取汇率）作为工具公开。然后，代理（甚至多个代理）可以使用标准化的模型上下文协议 (MCP) 访问这些工具。
FastMCP Python 软件包可用于创建公开名为 get_exchange_rate 的单个工具的 MCP 服务器。get_exchange_rate 工具通过互联网调用 Frankfurter API，以获取两种货币之间的当前汇率。
您可以在 mcp-server/server.py 文件中找到 MCP 服务器的代码：
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 服务器是否正常运行，以及是否可以使用模型上下文协议访问 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），以停止该服务器。
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 代理，以便在本地机器上创建通往远程 MCP 服务器的经过身份验证的隧道。
默认情况下，Cloud Run 服务的网址要求所有请求都必须通过 Cloud Run Invoker (roles/run.invoker) IAM 角色进行授权。此 IAM 政策绑定可确保使用强大的安全机制来验证本地 MCP 客户端的身份。
您应确保您或尝试访问远程 MCP 服务器的任何团队成员的 IAM 主账号（Google Cloud 账号）绑定了 roles/run.invoker IAM 角色。
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. 使用智能体开发套件 (ADK) 创建智能体
您已部署 MCP 服务器，现在可以使用智能体开发套件 (ADK) 创建币种代理。
智能体开发套件最近发布了 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 的开发者界面（通过运行 adk web 访问）：
uv run adk web
在浏览器中，前往 http://localhost:8000 查看并测试代理！
确保在 Web 界面的左上角选择了 currency_agent 作为代理。
在聊天区域中向代理提出问题，例如“250 加元兑换成美元是多少？”。您应该会看到代理在给出回答之前调用了我们的 get_exchange_rate MCP 工具。
代理有效！它可以处理与货币换算 💸 相关的问题。
7. Agent2Agent (A2A) 协议
Agent2Agent (A2A) 协议是一种开放标准，旨在实现 AI 代理之间的无缝通信和协作。这使得使用不同框架和由不同供应商构建的代理能够以通用语言相互通信，从而打破孤岛并促进互操作性。
A2A 可让代理执行以下操作：
- 发现：使用标准化的智能体卡片查找其他智能体并了解其技能 (AgentSkill) 和功能 (AgentCapabilities)。
- 通信：安全地交换消息和数据。
- 协作：委派任务并协调行动，以实现复杂的目标。
A2A 协议通过“代理卡片”等机制促进了这种通信，代理卡片充当数字名片，代理可以使用它来宣传自己的功能和连接信息。
现在，是时候使用 A2A 公开币种代理了，以便其他代理和客户端可以调用它。
A2A Python SDK
A2A Python SDK 为上述每种资源（AgentSkill、AgentCapabilities 和 AgentCard）提供了 Pydantic 模型。这提供了一个接口，用于加快开发速度并与 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 服务器
现在，您将查看一些代码片段，了解构成 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 作为 Web 框架，通常与 Uvicorn 等 ASGI 服务器一起运行。
# ... 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 服务器，使用智能体开发套件 (ADK) 创建了通过 MCP 连接到工具的币种代理，并使用 Agent2Agent (A2A) 协议公开了您的代理！货币代理现已可使用 A2A 与任何框架的其他代理进行交互！
点击此处可查看完整的代码文档。
所学内容
- 如何创建本地 MCP 服务器
- 将 MCP 服务器部署到 Cloud Run
- 如何使用智能体开发套件构建使用 MCP 工具的智能体
- 如何将 ADK 代理公开为 A2A 服务器
- 使用 A2A 客户端测试 A2A 服务器
清理
为避免系统因本实验中使用的资源向您的 Google Cloud 账号收取费用，请按照以下步骤操作：
- 在 Google Cloud 控制台中，前往管理资源页面。
- 在项目列表中，选择要删除的项目，然后点击删除。
- 在对话框中输入项目 ID，然后点击关停以删除项目。