1. 概览
AI 智能体正迅速受到欢迎，它们能够自主运行、学习并与环境互动以实现目标，从而彻底改变任务自动化和决策制定。
但具体来说，如何构建代理呢？此 Codelab 将向您展示如何构建一个可以转换不同国家/地区货币的货币代理，帮助您入门。本文旨在介绍最新技术，帮助您了解可能在网上看到的一些缩写词（MCP、ADK、A2A）。
Model Context Protocol (MCP)
Model Context Protocol (MCP) 是一种开放协议，可标准化应用向 LLM 提供上下文的方式。MCP 提供了一种将 AI 模型连接到资源、提示和工具的标准化方式。
智能体开发套件 (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 服务器，您可以编写轻量级程序，将特定功能（例如获取汇率）作为工具公开。然后，代理（甚至多个代理）可以使用标准化的 Model Context Protocol (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="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），以停止该服务器。
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.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 的开发者界面（通过运行 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],
)
现在，让我们将所有内容与货币代理结合起来，展示 A2A 的强大功能！
8. 公开币种代理 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 服务器，使用智能体开发套件 (ADK) 创建了使用 MCP 连接到工具的币种智能体，并使用 Agent2Agent (A2A) 协议公开了您的智能体！货币代理现已可使用 A2A 与任何框架的其他代理进行交互！
点击此处可查看完整的代码文档。
所学内容
- 如何创建本地 MCP 服务器
- 将 MCP 服务器部署到 Cloud Run
- 如何使用智能体开发套件构建使用 MCP 工具的智能体
- 如何将 ADK 代理作为 A2A 服务器公开
- 使用 A2A 客户端测试 A2A 服务器
清理
为避免系统因本实验中使用的资源向您的 Google Cloud 账号收取费用，请按照以下步骤操作：
- 在 Google Cloud 控制台中，前往管理资源页面。
- 在项目列表中，选择要删除的项目，然后点击删除。
- 在对话框中输入项目 ID，然后点击关停以删除项目。