Thông tin về lớp học lập trình này
1. Tổng quan
Các tác nhân AI đang ngày càng phổ biến, mang đến cuộc cách mạng trong việc tự động hoá tác vụ và đưa ra quyết định nhờ khả năng hoạt động độc lập, học hỏi và tương tác với môi trường để đạt được mục tiêu.
Nhưng chính xác thì làm cách nào để xây dựng một trợ lý ảo? Lớp học lập trình này sẽ giúp bạn bắt đầu bằng cách hướng dẫn cách tạo một tác nhân tiền tệ có thể chuyển đổi giữa các loại tiền tệ của nhiều quốc gia. Mục tiêu của khoá học này là giới thiệu các công nghệ mới nhất để giúp bạn hiểu rõ những từ viết tắt mà bạn có thể đã thấy trên Internet (MCP, ADK, A2A).
Giao thức bối cảnh mô hình (MCP)
Giao thức ngữ cảnh mô hình (MCP) là một giao thức mở giúp chuẩn hoá cách các ứng dụng cung cấp ngữ cảnh cho LLM. MCP cung cấp một cách chuẩn hoá để kết nối các mô hình AI với tài nguyên, câu lệnh và công cụ.
Agent Development Kit (ADK)
Agent Development Kit (ADK) là một khung điều phối linh hoạt để phát triển và triển khai các tác nhân AI. ADK không phụ thuộc vào mô hình, không phụ thuộc vào việc triển khai và được xây dựng để tương thích với các khung khác. ADK được thiết kế để giúp quá trình phát triển tác nhân giống với quá trình phát triển phần mềm hơn, giúp nhà phát triển dễ dàng tạo, triển khai và điều phối các cấu trúc dựa trên tác nhân, từ các tác vụ đơn giản đến quy trình công việc phức tạp.
Giao thức Agent2Agent (A2A)
Giao thức Agent2Agent (A2A) là một tiêu chuẩn mở được thiết kế để cho phép giao tiếp và cộng tác liền mạch giữa các tác nhân AI. Tương tự như cách MCP cung cấp một phương thức chuẩn hoá để cho phép các LLM truy cập vào dữ liệu và công cụ, A2A cung cấp một phương thức chuẩn hoá để các tác nhân giao tiếp với các tác nhân khác! Trong một thế giới mà các tác nhân được xây dựng bằng nhiều khung và bởi nhiều nhà cung cấp, A2A cung cấp một ngôn ngữ chung, phá vỡ các rào cản và thúc đẩy khả năng tương tác.
Kiến thức bạn sẽ học được
- Cách tạo máy chủ MCP cục bộ
- Triển khai máy chủ MCP lên Cloud Run
- Cách tạo một Đặc vụ bằng Bộ công cụ phát triển đặc vụ sử dụng các công cụ MCP
- Cách hiển thị một tác nhân ADK dưới dạng Máy chủ A2A
- Kiểm thử Máy chủ A2A bằng Ứng dụng A2A
Bạn cần có
2. Trước khi bắt đầu
Tạo một dự án
- Trong Google Cloud Console, trên trang chọn dự án, hãy chọn hoặc tạo một dự án trên Google Cloud.
- Đảm bảo bạn đã bật tính năng thanh toán cho dự án trên Cloud. Tìm hiểu cách kiểm tra xem tính năng thanh toán có được bật trên một dự án hay không.
- Kích hoạt Cloud Shell bằng cách nhấp vào đường liên kết này. Bạn có thể chuyển đổi giữa Cloud Shell Terminal (để chạy các lệnh trên đám mây) và Trình chỉnh sửa (để tạo dự án) bằng cách nhấp vào nút tương ứng trong Cloud Shell.
- Sau khi kết nối với Cloud Shell, bạn có thể kiểm tra để đảm bảo rằng bạn đã được xác thực và dự án được đặt thành mã dự án của bạn bằng lệnh sau:
gcloud auth list
- Chạy lệnh sau trong Cloud Shell để xác nhận rằng lệnh gcloud biết về dự án của bạn.
gcloud config list project
- Sử dụng lệnh sau để thiết lập dự án của bạn:
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- Bật các API bắt buộc bằng lệnh sau. Việc này có thể mất vài phút.
gcloud services enable cloudresourcemanager.googleapis.com \
servicenetworking.googleapis.com \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
- Đảm bảo bạn có Python 3.10 trở lên
Tham khảo tài liệu để biết các lệnh và cách sử dụng gcloud.
3. Cài đặt
- Sao chép kho lưu trữ:
git clone https://github.com/jackwotherspoon/currency-agent.git
cd currency-agent
- Cài đặt uv (dùng để quản lý các phần phụ thuộc):
# 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"
- Định cấu hình các biến môi trường (thông qua tệp
.env
):
Tạo tệp .env
bằng cách chạy lệnh sau:
echo "GOOGLE_GENAI_USE_VERTEXAI=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=us-central1" >> .env
4. Tạo một máy chủ MCP cục bộ
Trước khi điều phối tác nhân tiền tệ, trước tiên, bạn sẽ tạo một máy chủ MCP để hiển thị(các) công cụ mà tác nhân của bạn sẽ cần.
Máy chủ MCP cho phép bạn viết các chương trình đơn giản để cung cấp các chức năng cụ thể (chẳng hạn như tìm nạp tỷ giá hối đoái) dưới dạng công cụ. Sau đó, một hoặc thậm chí nhiều tác nhân có thể truy cập vào các công cụ này bằng Giao thức ngữ cảnh mô hình (MCP) được chuẩn hoá.
Bạn có thể tận dụng gói Python FastMCP để tạo một máy chủ MCP hiển thị một công cụ duy nhất có tên là get_exchange_rate
. Công cụ get_exchange_rate
gọi Frankfurter API qua Internet để lấy tỷ giá hối đoái hiện tại giữa hai loại tiền tệ.
Bạn có thể tìm thấy mã cho máy chủ MCP trong tệp 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),
)
)
Để khởi động máy chủ MCP cục bộ, hãy mở một cửa sổ dòng lệnh rồi chạy lệnh sau (máy chủ sẽ khởi động trên http://localhost:8080
):
uv run mcp-server/server.py
Kiểm tra để đảm bảo máy chủ MCP hoạt động đúng cách và bạn có thể truy cập vào công cụ get_exchange_rate
bằng Giao thức ngữ cảnh mô hình.
Trong cửa sổ dòng lệnh mới (để bạn không dừng máy chủ MCP cục bộ), hãy chạy lệnh sau:
uv run mcp-server/test_server.py
Bạn sẽ thấy tỷ giá hối đoái hiện tại của 1 USD (đô la Mỹ) sang EUR (Euro) được xuất ra:
--- 🛠️ 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
}
} ---
Tuyệt vời! Bạn đã có một máy chủ MCP hoạt động với một công cụ mà tác nhân của bạn có thể truy cập.
Trước khi chuyển sang trạm tiếp theo, hãy dừng máy chủ MCP đang chạy cục bộ bằng cách chạy Ctrl+C
(hoặc Command+C
trên máy Mac) trong thiết bị đầu cuối mà bạn đã khởi động.
5. Triển khai MCP Server lên Cloud Run
Giờ đây, bạn đã sẵn sàng triển khai máy chủ MCP dưới dạng máy chủ MCP từ xa cho Cloud Run 🚀☁️
Lợi ích của việc chạy máy chủ MCP từ xa
Việc chạy một máy chủ MCP từ xa trên Cloud Run có thể mang lại một số lợi ích:
- 📈Khả năng mở rộng: Cloud Run được thiết kế để mở rộng quy mô nhanh chóng nhằm xử lý tất cả các yêu cầu đến. Cloud Run sẽ tự động mở rộng quy mô máy chủ MCP dựa trên nhu cầu.
- 👥Máy chủ tập trung: Bạn có thể chia sẻ quyền truy cập vào máy chủ MCP tập trung với các thành viên trong nhóm thông qua đặc quyền IAM, cho phép họ kết nối với máy chủ đó từ máy cục bộ thay vì tất cả đều chạy máy chủ riêng của họ trên cục bộ. Nếu có thay đổi đối với máy chủ MCP, tất cả thành viên trong nhóm đều sẽ được hưởng lợi.
- 🔐Bảo mật: Cloud Run cung cấp một cách dễ dàng để buộc các yêu cầu được xác thực. Điều này chỉ cho phép các kết nối bảo mật đến máy chủ MCP của bạn, ngăn chặn hành vi truy cập trái phép.
Thay đổi thành thư mục mcp-server
:
cd mcp-server
Triển khai máy chủ MCP lên Cloud Run:
gcloud run deploy mcp-server --no-allow-unauthenticated --region=us-central1 --source .
Nếu dịch vụ của bạn đã triển khai thành công, bạn sẽ thấy một thông báo như sau:
Service [mcp-server] revision [mcp-server-12345-abc] has been deployed and is serving 100 percent of traffic.
Xác thực ứng dụng MCP
Vì bạn đã chỉ định --no-allow-unauthenticated
để yêu cầu xác thực, nên mọi ứng dụng MCP kết nối với máy chủ MCP từ xa đều cần xác thực.
Tài liệu chính thức về Lưu trữ máy chủ MCP trên Cloud Run cung cấp thêm thông tin về chủ đề này, tuỳ thuộc vào nơi bạn đang chạy ứng dụng MCP.
Bạn sẽ cần chạy proxy Cloud Run để tạo một đường hầm được xác thực đến máy chủ MCP từ xa trên máy cục bộ.
Theo mặc định, URL của các dịch vụ Cloud Run yêu cầu tất cả các yêu cầu phải được uỷ quyền bằng vai trò IAM Cloud Run Invoker (roles/run.invoker
). IAM policy binding này đảm bảo rằng một cơ chế bảo mật mạnh mẽ được dùng để xác thực ứng dụng MCP cục bộ của bạn.
Bạn nên đảm bảo rằng bạn hoặc bất kỳ thành viên nào trong nhóm đang cố gắng truy cập vào máy chủ MCP từ xa đều có vai trò roles/run.invoker
IAM được liên kết với nguyên tắc IAM (tài khoản Google Cloud) của họ.
gcloud run services proxy mcp-server --region=us-central1
Bạn sẽ thấy kết quả sau đây:
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
Giờ đây, tất cả lưu lượng truy cập đến http://127.0.0.1:8080
sẽ được xác thực và chuyển tiếp đến máy chủ MCP từ xa.
Kiểm thử máy chủ MCP từ xa
Trong thiết bị đầu cuối mới, hãy quay lại thư mục gốc và chạy lại tệp mcp-server/test_server.py
để đảm bảo máy chủ MCP từ xa đang hoạt động.
cd ..
uv run mcp-server/test_server.py
Bạn sẽ thấy kết quả tương tự như khi chạy máy chủ cục bộ:
--- 🛠️ 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
}
} ---
Bạn có thể truy vấn nhật ký của máy chủ MCP Cloud Run đã triển khai nếu muốn xác minh rằng máy chủ từ xa thực sự đã được gọi:
gcloud run services logs read mcp-server --region us-central1 --limit 5
Bạn sẽ thấy kết quả sau đây trong nhật ký:
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}}
Giờ đây, bạn đã có một máy chủ MCP từ xa, bạn có thể chuyển sang tạo một tác nhân! 🤖
6. Tạo một nhân viên hỗ trợ bằng Agent Development Kit (ADK)
Bạn đã triển khai máy chủ MCP, giờ là lúc tạo tác nhân tiền tệ bằng Bộ phát triển tác nhân (ADK).
Agent Development Kit gần đây đã phát hành bản phát hành ổn định v1.0.0. Cột mốc này cho thấy Python ADK hiện đã sẵn sàng cho phiên bản phát hành công khai, mang đến một nền tảng đáng tin cậy và mạnh mẽ để nhà phát triển tự tin xây dựng và triển khai các tác nhân của họ trong môi trường thực.
ADK giúp tạo các tác nhân cực kỳ gọn nhẹ và cho phép các tác nhân dễ dàng kết nối với máy chủ MCP nhờ tính năng hỗ trợ tích hợp cho MCP Tools. Tác nhân tiền tệ sẽ truy cập vào công cụ get_exchange_rate
bằng cách sử dụng lớp MCPToolset của ADK.
Mã của tác nhân tiền tệ nằm trong 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()
Để nhanh chóng kiểm thử tác nhân tiền tệ, bạn có thể tận dụng giao diện người dùng dành cho nhà phát triển của ADK bằng cách chạy adk web
:
uv run adk web
Trong trình duyệt, hãy truy cập vào http://localhost:8000
để xem và thử nghiệm tác nhân!
Đảm bảo rằng currency_agent
được chọn làm tác nhân ở góc trên bên trái của giao diện người dùng web.
Hỏi nhân viên hỗ trợ trong khu vực trò chuyện, chẳng hạn như "250 CAD bằng bao nhiêu USD?". Bạn sẽ thấy nhân viên gọi công cụ get_exchange_rate
MCP của chúng tôi trước khi đưa ra câu trả lời.
Tác nhân hoạt động! Công cụ này có thể xử lý các câu hỏi xoay quanh việc chuyển đổi tiền tệ 💸.
7. Giao thức Agent2Agent (A2A)
Giao thức Agent2Agent (A2A) là một tiêu chuẩn mở được thiết kế để cho phép giao tiếp và cộng tác liền mạch giữa các tác nhân AI. Điều này cho phép các tác nhân được xây dựng bằng nhiều khung và bởi nhiều nhà cung cấp khác nhau giao tiếp với nhau bằng một ngôn ngữ chung, phá vỡ các rào cản và thúc đẩy khả năng tương tác.
A2A cho phép nhân viên hỗ trợ:
- Khám phá: Tìm các tác nhân khác và tìm hiểu kỹ năng (AgentSkill) và khả năng (AgentCapabilities) của họ bằng cách sử dụng Thẻ tác nhân tiêu chuẩn.
- Liên lạc: Trao đổi tin nhắn và dữ liệu một cách an toàn.
- Cộng tác: Uỷ quyền nhiệm vụ và phối hợp hành động để đạt được các mục tiêu phức tạp.
Giao thức A2A hỗ trợ hoạt động giao tiếp này thông qua các cơ chế như "Thẻ đại lý". Đây là thẻ doanh nghiệp kỹ thuật số mà các đại lý có thể dùng để quảng cáo khả năng và thông tin kết nối của họ.
Bây giờ là lúc bạn cần hiển thị tác nhân tiền tệ bằng cách sử dụng A2A để các tác nhân và ứng dụng khác có thể gọi tác nhân này.
A2A Python SDK
A2A Python SDK cung cấp các mô hình Pydantic cho từng tài nguyên nêu trên; AgentSkill, AgentCapabilities và AgentCard. Điều này cung cấp một giao diện để đẩy nhanh quá trình phát triển và tích hợp với giao thức A2A.
AgentSkill
là cách bạn quảng cáo cho các tác nhân khác rằng tác nhân tiền tệ có một công cụ để 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?'],
)
Sau đó, trong phần AgentCard
, hệ thống sẽ liệt kê các kỹ năng và khả năng của trợ lý ảo cùng với các thông tin chi tiết khác như chế độ đầu vào và đầu ra mà trợ lý ảo có thể xử lý:
# 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],
)
Giao diện AgentExecutor xử lý logic cốt lõi về cách một tác nhân A2A xử lý các yêu cầu và tạo ra các phản hồi/sự kiện. A2A Python SDK cung cấp một lớp cơ sở trừu tượng a2a.server.agent_execution.AgentExecutor
mà bạn cần triển khai.
Đã đến lúc kết hợp tất cả với tác nhân tiền tệ và thể hiện sức mạnh của A2A!
8. Máy chủ A2A của Currency Agent
Bây giờ, bạn sẽ xem xét một số đoạn mã và xem cách các đoạn mã khác nhau tạo nên một máy chủ A2A.
Khi xem bên trong tệp currency_agent/agent_executor.py
, bạn sẽ thấy lớp ADKAgentExecutor
kế thừa từ lớp AgentExecutor
trừu tượng A2A. Nền tảng này xử lý việc gọi tác nhân ADK bằng cách gọi trình chạy ADK, xử lý các yêu cầu gửi đến tác nhân và chuyển đổi qua lại giữa google.genai.types
mà ADK sử dụng và a2a.types
mà A2A sử dụng.
# ... 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
Bên trong currency_agent/__main__.py
là nơi bạn khởi tạo AgentSkill, AgentCard và tạo tác nhân tiền tệ ADK. Đây cũng là nơi bạn thiết lập và khởi động máy chủ A2A.
A2A Python SDK cung cấp một lớp A2AFastAPIApplication
giúp đơn giản hoá việc chạy một máy chủ HTTP tuân thủ A2A. Nó sử dụng FastAPI cho khung web và thường chạy với một máy chủ ASGI như 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
Để chạy máy chủ A2A, trong cửa sổ dòng lệnh mới, hãy chạy lệnh sau:
uv run currency_agent/
Nếu máy chủ khởi động thành công, đầu ra sẽ có dạng như sau, cho biết máy chủ đang chạy trên cổng 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)
Giờ đây, tác nhân tiền tệ đã chạy thành công dưới dạng một máy chủ A2A, có khả năng được các tác nhân hoặc ứng dụng khác gọi bằng giao thức A2A!
Kiểm thử Máy chủ A2A
Giờ đây, bạn có thể kiểm thử máy chủ bằng cách gửi một số yêu cầu cho máy chủ đó thông qua A2A!
A2A Python SDK cung cấp một lớp a2a.client.A2AClient
giúp bạn đơn giản hoá quy trình này.
Tệp currency_agent/test_client.py
chứa mã chạy qua một số trường hợp kiểm thử khác nhau đối với máy chủ 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.')
Chạy các kiểm thử bằng lệnh sau:
uv run currency_agent/test_client.py
Một lần chạy thử nghiệm thành công sẽ dẫn đến những kết quả sau:
--- 🔄 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. ---
Ứng dụng hoạt động rồi! Bạn đã thử nghiệm thành công việc có thể giao tiếp với tác nhân tiền tệ thông qua máy chủ A2A! 🎉
Hãy xem kho lưu trữ a2a-samples trên GitHub để xem các trường hợp sử dụng nâng cao hơn!
Bạn muốn triển khai tác nhân của mình? Vertex AI Agent Engine mang đến trải nghiệm được quản lý để triển khai các tác nhân AI vào hoạt động thực tế!
9. Xin chúc mừng
Xin chúc mừng! Bạn đã tạo và triển khai thành công một máy chủ MCP từ xa, tạo một tác nhân tiền tệ bằng Bộ công cụ phát triển tác nhân (ADK) kết nối với các công cụ bằng MCP và hiển thị tác nhân của bạn bằng giao thức Agent2Agent (A2A)! Giờ đây, nhân viên hỗ trợ tiền tệ có thể tương tác với các nhân viên hỗ trợ khác thuộc mọi khung hình bằng cách sử dụng A2A!
Tại đây là đường liên kết đến tài liệu đầy đủ về mã.
Nội dung đã đề cập
- Cách tạo máy chủ MCP cục bộ
- Triển khai máy chủ MCP lên Cloud Run
- Cách tạo một Đặc vụ bằng Bộ công cụ phát triển đặc vụ sử dụng các công cụ MCP
- Cách hiển thị một tác nhân ADK dưới dạng Máy chủ A2A
- Kiểm thử Máy chủ A2A bằng Ứng dụng A2A
Dọn dẹp
Để tránh phát sinh phí cho tài khoản Google Cloud của bạn đối với các tài nguyên được dùng trong phòng thí nghiệm này, hãy làm theo các bước sau:
- Trong Google Cloud Console, hãy chuyển đến trang Quản lý tài nguyên.
- Trong danh sách dự án, hãy chọn dự án mà bạn muốn xoá, rồi nhấp vào Xoá.
- Trong hộp thoại, hãy nhập mã dự án rồi nhấp vào Tắt để xoá dự án.