لمحة عن هذا الدرس التطبيقي حول الترميز
1. نظرة عامة
تتزايد شعبية وكلاء الذكاء الاصطناعي بسرعة، إذ أحدثوا ثورة في أتمتة المهام واتخاذ القرارات بفضل قدرتهم على العمل بشكل مستقل والتعلم والتفاعل مع بيئتهم لتحقيق الأهداف.
ولكن كيف يمكن إنشاء وكيل؟ سيساعدك هذا الدرس العملي على البدء من خلال توضيح كيفية إنشاء وكيل عملة يمكنه التحويل بين عملات البلدان المختلفة. هدفنا هو شرح أحدث التقنيات لمساعدتك في فهم الاختصارات التي ربما رأيتها منتشرة على الإنترنت (مثل MCP وADK وA2A).
بروتوكول سياق النموذج (MCP)
بروتوكول سياق النموذج (MCP) هو بروتوكول مفتوح يوحّد طريقة تقديم التطبيقات للسياق إلى النماذج اللغوية الكبيرة. توفّر MCP طريقة موحّدة لربط نماذج الذكاء الاصطناعي بالمراجع والطلبات والأدوات.
حزمة تطوير البرامج (ADK) الخاصة بالوكلاء
مجموعة تطوير البرامج (ADK) هي إطار عمل مرن لتنظيم عملية تطوير ونشر وكلاء الذكاء الاصطناعي. لا يعتمد ADK على نموذج معيّن أو عملية نشر معيّنة، وهو مصمّم ليتوافق مع أُطر العمل الأخرى. تم تصميم ADK لجعل عملية تطوير الوكيل تبدو أقرب إلى تطوير البرامج، وذلك لتسهيل إنشاء المطوّرين ونشرهم وتنسيقهم للبُنى الأساسية المستندة إلى الوكلاء التي تتراوح بين المهام البسيطة وسير العمل المعقّد.
بروتوكول Agent2Agent (A2A)
بروتوكول Agent2Agent (A2A) هو معيار مفتوح مصمّم لإتاحة التواصل والتعاون السلسَين بين نماذج الذكاء الاصطناعي الوكيلة. تمامًا مثلما توفّر منصة MCP طريقة موحَّدة لمنح نماذج اللغات الكبيرة إمكانية الوصول إلى البيانات والأدوات، توفّر منصة A2A طريقة موحَّدة للتواصل بين الوكلاء. في عالم يتم فيه إنشاء العملاء باستخدام أُطر متنوعة ومن قِبل مورّدين مختلفين، توفّر A2A لغة مشتركة، ما يؤدي إلى إزالة الحواجز وتعزيز قابلية التشغيل التفاعلي.
المُعطيات
- كيفية إنشاء خادم MCP محلي
- نشر خادم MCP على Cloud Run
- كيفية إنشاء وكيل باستخدام "حزمة تطوير الوكلاء" التي تستخدم أدوات MCP
- كيفية عرض وكيل ADK كخادم A2A
- اختبار خادم A2A باستخدام عميل A2A
المتطلبات
2. قبل البدء
إنشاء مشروع
- في Google Cloud Console، في صفحة اختيار المشروع، اختَر مشروعًا على Google Cloud أو أنشِئ مشروعًا.
- تأكَّد من تفعيل الفوترة لمشروعك على Cloud. كيفية التحقّق من تفعيل الفوترة في مشروع
- فعِّل Cloud Shell من خلال النقر على هذا الرابط. يمكنك التبديل بين "نافذة Cloud Shell" (لتنفيذ أوامر السحابة الإلكترونية) و"المحرّر" (لإنشاء المشاريع) من خلال النقر على الزر المناسب من Cloud Shell.
- بعد الاتصال بـ Cloud Shell، يمكنك التأكّد من أنّك قد تم التحقّق من هويتك وأنّه تم ضبط المشروع على معرّف مشروعك باستخدام الأمر التالي:
gcloud auth list
- نفِّذ الأمر التالي في Cloud Shell للتأكّد من أنّ أمر gcloud يعرف مشروعك.
gcloud config list project
- استخدِم الأمر التالي لضبط مشروعك:
export PROJECT_ID=<YOUR_PROJECT_ID>
gcloud config set project $PROJECT_ID
- فعِّل واجهات برمجة التطبيقات المطلوبة باستخدام الأمر التالي. قد يستغرق هذا بضع دقائق.
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 أو الإصدارات الأحدث من Python
يمكنك الرجوع إلى المستندات لمعرفة أوامر 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 لإنشاء خادم 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 يعمل بشكل صحيح وما إذا كان يمكن الوصول إلى الأداة get_exchange_rate باستخدام بروتوكول سياق النموذج.
في نافذة وحدة طرفية جديدة (حتى لا توقف خادم MCP المحلي)، شغِّل ما يلي:
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 يعمل مع أداة يمكن أن يصل إليها وكيلك.
قبل الانتقال إلى المحطة التالية، أوقِف خادم MCP الذي يتم تشغيله محليًا عن طريق تنفيذ Ctrl+C (أو Command+C على جهاز Mac) في الوحدة الطرفية التي بدأت منها.
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.
عليك تشغيل خادم وكيل Cloud Run لإنشاء نفق مصادَق عليه إلى خادم MCP البعيد على جهازك المحلي.
بشكلٍ تلقائي، يتطلّب عنوان URL لخدمات Cloud Run أن يتمّ تفويض جميع الطلبات باستخدام دور منفّذ Cloud Run (roles/run.invoker) في "إدارة الهوية وإمكانية الوصول". يضمن ربط سياسة إدارة الهوية وإمكانية الوصول (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
}
} ---
يمكنك طلب البحث في سجلات خادم 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).
تم مؤخرًا طرح الإصدار الثابت 1.0.0 من Agent Development Kit. يشير هذا الإنجاز إلى أنّ حزمة تطوير التطبيقات (ADK) من Python أصبحت الآن جاهزة للإنتاج، ما يوفّر للمطوّرين منصة موثوقة وقوية تتيح لهم إنشاء وكلاء ونشرهم بثقة في البيئات المباشرة.
تتيح "حزمة تطوير البرامج" إنشاء وكلاء خفيفي الوزن للغاية، كما تتيح لهم الاتصال بسهولة بخوادم MCP مع توفير دعم مضمّن لأدوات MCP. سيصل وكيل العملة إلى الأداة get_exchange_rate باستخدام فئة MCPToolset في حزمة تطوير البرامج الإعلانية (ADK).
يقع رمز وكيل العملة في 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 دولار كندي بالدولار الأمريكي؟". من المفترض أن ترى الموظف يتصل بأداة get_exchange_rate MCP قبل أن تقدّم الأداة ردًا.
يعمل الوكيل بشكل جيد. يمكنه التعامل مع طلبات البحث التي تدور حول تحويل العملات 💸.
7. بروتوكول Agent2Agent (A2A)
بروتوكول Agent2Agent (A2A) هو معيار مفتوح مصمّم لإتاحة التواصل والتعاون السلسَين بين نماذج الذكاء الاصطناعي الوكيلة. يتيح ذلك للوكلاء الذين تم إنشاؤهم باستخدام أُطر متنوعة ومن قِبل مورّدين مختلفين التواصل مع بعضهم البعض بلغة مشتركة، ما يؤدي إلى إزالة الحواجز وتعزيز قابلية التشغيل التفاعلي.
تتيح ميزة "التطبيقات إلى التطبيقات" للوكلاء ما يلي:
- الاستكشاف: يمكنك العثور على وكلاء آخرين والتعرّف على مهاراتهم (AgentSkill) وقدراتهم (AgentCapabilities) باستخدام بطاقات الوكيل الموحّدة.
- التواصل: تبادل الرسائل والبيانات بأمان
- التعاون: تفويض المهام وتنسيق الإجراءات لتحقيق أهداف معقّدة
يسهّل بروتوكول A2A عملية التواصل هذه من خلال آليات مثل "بطاقات الوكيل" التي تعمل كبطاقات عمل رقمية يمكن للوكلاء استخدامها للإعلان عن إمكاناتهم ومعلومات الاتصال الخاصة بهم.
حان الوقت الآن لعرض وكيل العملة باستخدام A2A حتى تتمكّن الوكلاء والعملاء الآخرون من استخدامه.
A2A Python SDK
توفّر حزمة تطوير البرامج (SDK) الخاصة بواجهة A2A في Python نماذج Pydantic لكل من الموارد المذكورة أعلاه: AgentSkill وAgentCapabilities و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 للطلبات وإنشاء الردود/الأحداث. توفّر حزمة تطوير البرامج (SDK) الخاصة بتطبيق A2A بلغة Python فئة أساسية مجرّدة a2a.server.agent_execution.AgentExecutor يجب تنفيذها.
حان الوقت الآن لربط كل ذلك معًا باستخدام وكيل العملة وعرض قوة ميزة "التطبيق إلى التطبيق"!
8. خادم Currency Agent A2A
سنلقي الآن نظرة على بعض أجزاء الرمز البرمجي ونرى كيف تتكامل الأجزاء المختلفة التي تشكّل خادمًا من النوع A2A.
عند إلقاء نظرة داخل الملف currency_agent/agent_executor.py، ستجد الصف ADKAgentExecutor الذي يرث من الفئة المجردة A2A 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 وبدء استخدامه.
توفر حزمة تطوير البرامج (SDK) الخاصة بلغة Python في A2A فئة 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.
توفّر حزمة تطوير البرامج (SDK) للغة Python من A2A فئة 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-samples على GitHub للتعرّف على المزيد من حالات الاستخدام المتقدّمة.
هل تريد نشر الوكيل؟ توفّر Vertex AI Agent Engine تجربة مُدارة لنشر وكلاء الذكاء الاصطناعي في مرحلة الإنتاج.
9. تهانينا
تهانينا! لقد أنشأت ونشرت بنجاح خادم MCP عن بُعد، وأنشأت وكيل عملة باستخدام "حزمة تطوير الوكيل" (ADK) التي تتصل بالأدوات باستخدام MCP، وعرضت الوكيل باستخدام بروتوكول Agent2Agent (A2A). أصبح وكيل العملة متاحًا الآن للتفاعل مع وكلاء آخرين من أي إطار عمل باستخدام ميزة "التطبيق إلى التطبيق" (A2A).
إليك رابط يؤدي إلى مستندات الرمز البرمجي الكاملة.
المواضيع التي تناولناها
- كيفية إنشاء خادم MCP محلي
- نشر خادم MCP على Cloud Run
- كيفية إنشاء وكيل باستخدام "حزمة تطوير الوكلاء" التي تستخدم أدوات MCP
- كيفية عرض وكيل ADK كخادم A2A
- اختبار خادم A2A باستخدام عميل A2A
تَنظيم
لتجنُّب تحمّل رسوم في حسابك على Google Cloud مقابل الموارد المستخدَمة في هذا المختبر، اتّبِع الخطوات التالية:
- في Google Cloud Console، انتقِل إلى صفحة إدارة الموارد.
- في قائمة المشاريع، اختَر المشروع الذي تريد حذفه، ثم انقر على حذف.
- في مربّع الحوار، اكتب رقم تعريف المشروع، ثم انقر على إيقاف لحذف المشروع.