1. بررسی اجمالی
محبوبیت عوامل هوش مصنوعی به سرعت در حال افزایش است و اتوماسیون وظایف و تصمیم‌گیری را با توانایی خود برای عملکرد مستقل، یادگیری و تعامل با محیط خود برای دستیابی به اهداف متحول می‌کند.
اما دقیقاً چگونه می توان در مورد ساخت یک نماینده اقدام کرد؟ این کد لبه با نشان دادن نحوه ساخت یک عامل ارزی که می تواند بین ارزهای کشورهای مختلف تبدیل شود، به شما کمک می کند تا شروع کنید. با هدف قدم زدن در میان آخرین فن آوری ها برای کمک به درک کلمات اختصاری که ممکن است در اینترنت مشاهده کرده باشید (MCP، ADK، A2A).
پروتکل بافت مدل (MCP)
پروتکل بافت مدل (MCP) یک پروتکل باز است که نحوه ارائه زمینه برنامه‌ها به LLM را استاندارد می‌کند. MCP یک راه استاندارد برای اتصال مدل‌های هوش مصنوعی به منابع، درخواست‌ها و ابزارها ارائه می‌کند.
کیت توسعه عامل (ADK)
کیت توسعه عامل (ADK) یک چارچوب ارکستراسیون منعطف برای توسعه و استقرار عوامل هوش مصنوعی است. ADK مدل-agnostic، استقرار-agnostic است و برای سازگاری با سایر چارچوب ها ساخته شده است. ADK طوری طراحی شده است که توسعه عامل را بیشتر شبیه به توسعه نرم افزار کند، تا توسعه دهندگان را برای ایجاد، استقرار و هماهنگی معماری های عاملی که از وظایف ساده تا گردش های کاری پیچیده را شامل می شود، آسان تر کند.
پروتکل Agent2Agent (A2A).
پروتکل Agent2Agent (A2A) یک استاندارد باز است که برای برقراری ارتباط و همکاری یکپارچه بین عوامل هوش مصنوعی طراحی شده است. درست مانند روشی که MCP یک راه استاندارد برای دسترسی LLMها به داده ها و ابزارها را فراهم می کند، A2A نیز یک راه استاندارد برای صحبت نمایندگان با سایر عوامل ارائه می دهد! در دنیایی که عوامل با استفاده از چارچوب‌های متنوع و توسط فروشندگان مختلف ساخته می‌شوند، A2A زبان مشترکی را فراهم می‌کند، سیلوها را خراب می‌کند و قابلیت همکاری را تقویت می‌کند.
چیزی که یاد خواهید گرفت
- نحوه ایجاد یک سرور MCP محلی
- استقرار سرور MCP در Cloud Run
- نحوه ساخت یک عامل با کیت توسعه عامل که از ابزارهای MCP استفاده می کند
- نحوه افشای یک عامل ADK به عنوان یک سرور A2A
- تست سرور A2A با استفاده از A2A Client
آنچه شما نیاز دارید
2. قبل از شروع
یک پروژه ایجاد کنید
- در Google Cloud Console ، در صفحه انتخاب پروژه، یک پروژه Google Cloud را انتخاب یا ایجاد کنید.
- مطمئن شوید که صورتحساب برای پروژه Cloud شما فعال است. با نحوه بررسی فعال بودن صورتحساب در پروژه آشنا شوید.
- Cloud Shell را با کلیک بر روی این پیوند فعال کنید. می توانید با کلیک بر روی دکمه مربوطه از Cloud Shell بین Cloud Shell Terminal (برای اجرای دستورات ابری) و Editor (برای ساخت پروژه ها) جابه جا شوید.
- پس از اتصال به 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
- حتما پایتون 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) به این ابزارها دسترسی داشته باشند.
بسته Python FastMCP را می توان برای ایجاد یک سرور MCP استفاده کرد که یک ابزار واحد به نام get_exchange_rate در معرض دید قرار می دهد. ابزار get_exchange_rate از طریق اینترنت با فرانکفورتر 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 به درستی کار می کند و ابزار get_exchange_rate با استفاده از پروتکل بافت مدل قابل دسترسی است.
در یک پنجره ترمینال جدید (به طوری که سرور MCP محلی را متوقف نکنید) موارد زیر را اجرا کنید:
uv run mcp-server/test_server.py
شما باید نرخ فعلی مبادله 1 USD (دلار آمریکا) به یورو (یورو) را مشاهده کنید:
--- 🛠️ 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 در حال کار با ابزاری دارید که نماینده شما می‌تواند به آن دسترسی داشته باشد.
قبل از رفتن به ایستگاه بعدی، با اجرای Ctrl+C (یا Command+C در Mac) در ترمینالی که آن را راه‌اندازی کرده‌اید، سرور MCP را که به صورت محلی اجرا می‌شود، متوقف کنید.
5. سرور MCP خود را در Cloud Run مستقر کنید
اکنون آماده استقرار سرور MCP به عنوان یک سرور MCP راه دور در Cloud Run هستید 🚀☁️
مزایای اجرای سرور MCP از راه دور
اجرای یک سرور MCP از راه دور در Cloud Run می تواند چندین مزیت داشته باشد:
- 📈مقیاس‌پذیری : Cloud Run به گونه‌ای ساخته شده است که به سرعت برای رسیدگی به تمام درخواست‌های دریافتی گسترش یابد . Cloud Run سرور MCP شما را به طور خودکار بر اساس تقاضا مقیاس می دهد.
- 👥سرور متمرکز : می توانید دسترسی به یک سرور MCP متمرکز را با اعضای تیم از طریق امتیازات IAM به اشتراک بگذارید و به آنها اجازه دهید به جای اینکه همه سرورهای خود را به صورت محلی اجرا کنند، از ماشین های محلی خود به آن متصل شوند. اگر تغییری در سرور 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 راه دور متصل می شود باید احراز هویت شود.
اسناد رسمی سرورهای میزبان MCP در Cloud Run بسته به جایی که مشتری MCP خود را اجرا می کنید، اطلاعات بیشتری در مورد این موضوع ارائه می دهد.
برای ایجاد یک تونل احراز هویت شده به سرور MCP راه دور در دستگاه محلی خود، باید پروکسی Cloud Run را اجرا کنید.
به‌طور پیش‌فرض، URL سرویس‌های Cloud Run نیاز دارد که همه درخواست‌ها با نقش IAM Cloud Run Invoker ( roles/run.invoker ) مجاز شوند. این الزام‌آوری خط‌مشی IAM تضمین می‌کند که یک مکانیسم امنیتی قوی برای تأیید اعتبار مشتری MCP محلی شما استفاده می‌شود.
باید مطمئن شوید که شما یا هر یک از اعضای تیمی که سعی می‌کنند به سرور MCP راه دور دسترسی پیدا کنند، نقش IAM roles/run.invoker به اصل 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
}
} ---
اگر می‌خواهید تأیید کنید که سرور راه دور واقعاً نامیده شده است، می‌توانید گزارش‌های سرور MCP Cloud Run مستقر شده را پرس و جو کنید:
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) است.
Agent Development Kit اخیرا نسخه پایدار نسخه 1.0.0 خود را منتشر کرده است. این نقطه عطف نشان می‌دهد که Python ADK اکنون آماده تولید است و یک پلتفرم قابل اعتماد و قوی برای توسعه‌دهندگان ارائه می‌کند تا با اطمینان عوامل خود را در محیط‌های زنده بسازند و مستقر کنند.
ADK ایجاد عامل ها را بسیار سبک می کند و به آنها اجازه می دهد تا به راحتی به سرورهای MCP با پشتیبانی داخلی از ابزار MCP متصل شوند. عامل ارز با استفاده از کلاس MCPToolset ADK به ابزار 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 بروید تا عامل را ببینید و آزمایش کنید!
مطمئن شوید که currency_agent به عنوان نماینده در گوشه سمت چپ بالای رابط کاربری وب انتخاب شده است.
از نماینده خود در منطقه چت چیزی مانند "250 CAD به USD چیست؟" . باید ببینید که عامل قبل از اینکه پاسخی بدهد، ابزار get_exchange_rate MCP ما را صدا می کند.
نماینده کار می کند! می تواند پرس و جوهایی را که حول تبدیل ارز می چرخند رسیدگی کند 💸.
7. پروتکل Agent2Agent (A2A).
پروتکل Agent2Agent (A2A) یک استاندارد باز است که برای برقراری ارتباط و همکاری یکپارچه بین عوامل هوش مصنوعی طراحی شده است. این به عواملی که با استفاده از چارچوب‌های متنوع و توسط فروشندگان مختلف ساخته می‌شوند، اجازه می‌دهد تا با یکدیگر به یک زبان مشترک ارتباط برقرار کنند، سیلوها را شکسته و قابلیت همکاری را تقویت کنند.
A2A به نمایندگان اجازه می دهد:
- کشف کنید: عوامل دیگر را بیابید و مهارت‌ها ( AgentSkill ) و قابلیت‌های ( AgentCapabilities ) آنها را با استفاده از کارت‌های عامل استاندارد بیاموزید.
- برقراری ارتباط: پیام ها و داده ها را به صورت ایمن مبادله کنید.
- همکاری: واگذاری وظایف و هماهنگ کردن اقدامات برای دستیابی به اهداف پیچیده.
پروتکل A2A این ارتباط را از طریق مکانیسم‌هایی مانند «کارت‌های عامل» که به‌عنوان نماینده کارت‌های ویزیت دیجیتالی عمل می‌کنند، برای تبلیغ قابلیت‌ها و اطلاعات اتصال خود استفاده می‌کنند، تسهیل می‌کند.
اکنون زمان افشای عامل ارز با استفاده از A2A است تا توسط سایر نمایندگان و مشتریان فراخوانی شود.
A2A Python SDK
A2A Python SDK مدل های Pydantic را برای هر یک از منابع فوق الذکر ارائه می دهد. AgentSkill ، AgentCabilities و 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],
)
رابط AgentExecutor منطق اصلی نحوه پردازش درخواست‌ها و تولید پاسخ‌ها/رویدادها توسط یک عامل A2A را مدیریت می‌کند. A2A Python SDK یک کلاس پایه انتزاعی a2a.server.agent_execution.AgentExecutor را ارائه می دهد که باید آن را پیاده سازی کنید.
زمان آن فرا رسیده است که همه چیز را با عامل ارز کنار هم بگذاریم و قدرت A2A را به نمایش بگذاریم!
8. عامل ارز سرور A2A
اکنون به برخی از کدها نگاهی می اندازید و خواهید دید که چگونه قطعات مختلفی که یک سرور A2A را تشکیل می دهند با هم ترکیب می شوند.
نگاهی به داخل فایل currency_agent/agent_executor.py جایی است که کلاس ADKAgentExecutor را دارید که از کلاس A2A abstract AgentExecutor به ارث می‌برد. فراخوانی عامل ADK را با فراخوانی اجراکننده ADK، پردازش درخواست ها به عامل، و تبدیل بین google.genai.types که ADK استفاده می کند و a2a.types که A2A استفاده می کند، انجام می دهد.
# ... 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 ارائه می کند که اجرای یک سرور HTTP سازگار با A2A را ساده می کند. از 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 ارتباط برقرار کنید! 🎉
برای مشاهده موارد استفاده پیشرفته تر ، مخزن نمونه a2a را در GitHub بررسی کنید!
به دنبال استقرار عامل خود هستید؟ Vertex AI Agent Engine یک تجربه مدیریت شده برای استقرار عوامل هوش مصنوعی در تولید فراهم می کند!
9. تبریک می گویم
تبریک می گویم! شما با موفقیت یک سرور MCP راه دور ساخته و مستقر کرده اید، یک عامل ارز با استفاده از کیت توسعه عامل (ADK) ایجاد کرده اید که با استفاده از MCP به ابزارها متصل می شود، و عامل خود را با استفاده از پروتکل Agent2Agent (A2A) در معرض نمایش قرار داده اید! عامل ارز اکنون برای تعامل با سایر عوامل از هر چارچوبی با استفاده از A2A در دسترس است!
در اینجا یک پیوند به مستندات کد کامل است.
آنچه را پوشش داده ایم
- نحوه ایجاد یک سرور MCP محلی
- استقرار سرور MCP در Cloud Run
- نحوه ساخت یک عامل با کیت توسعه عامل که از ابزارهای MCP استفاده می کند
- نحوه افشای یک عامل ADK به عنوان یک سرور A2A
- تست سرور A2A با استفاده از A2A Client
پاک کن
برای جلوگیری از تحمیل هزینه به حساب Google Cloud خود برای منابع مورد استفاده در این آزمایشگاه، این مراحل را دنبال کنید:
- در کنسول Google Cloud، به صفحه مدیریت منابع بروید.
- در لیست پروژه، پروژه ای را که می خواهید حذف کنید انتخاب کنید و سپس روی Delete کلیک کنید.
- در محاوره، شناسه پروژه را تایپ کنید و سپس روی Shut down کلیک کنید تا پروژه حذف شود.