بدء استخدام MCP وA2A مع ADK

1. نظرة عامة

تزداد شعبية وكلاء الذكاء الاصطناعي بسرعة، ما يُحدث ثورة في أتمتة المهام واتخاذ القرارات بفضل قدرتهم على العمل بشكل مستقل والتعلّم والتفاعل مع بيئتهم لتحقيق الأهداف.

سيساعدك هذا الدرس التطبيقي حول الترميز في البدء من خلال إنشاء نظام متكامل متعدد الوكلاء. ستنشئ خادم MCP للعملات، ووكيل عملات يستخدم خادم MCP ويتم عرضه عبر بروتوكول Agent2Agent (A2A)، ووكيل سفر يتضمّن وكيل طقس فرعي يستخدم وكيل العملات عبر A2A، وأخيرًا ستنشر جميع الخدمات على Cloud Run.

هل تفضّل استخدام Java؟ يتناول هذا الدرس التطبيقي حول الترميز عملية التنفيذ بلغة Python، ولكن يتوفّر أيضًا تنفيذ كامل بلغة Java باستخدام ADK للغة Java وA2A Java SDK وMCP Java SDK في المجلد java/ من المستودع.

Core Technologies

بروتوكول سياق النموذج (MCP)

بروتوكول سياق النموذج (MCP) هو بروتوكول مفتوح يوحّد طريقة توفير التطبيقات للسياق والأدوات للنماذج اللغوية الكبيرة. يمكنك اعتبار بروتوكول MCP مثل منفذ USB-C لتطبيقات الذكاء الاصطناعي: فهو يوفّر طريقة موحّدة لربط نماذج الذكاء الاصطناعي بمصادر البيانات الخارجية وواجهات برمجة التطبيقات والأدوات.

في هذا الدرس التطبيقي حول الترميز، يعرض currency_mcp_server أداة get_exchange_rate عبر Streamable HTTP (/mcp)، مستندة إلى Frankfurter API العامة.

حزمة تطوير الوكلاء (ADK)

حزمة تطوير الوكلاء (ADK) هي إطار عمل مرن لتنسيق وتطوير ونشر وكلاء الذكاء الاصطناعي. لا تعتمد حزمة ADK على نموذج أو طريقة نشر معيّنة، وهي مصمّمة لتتوافق مع إطارات العمل الأخرى. تسهّل حزمة ADK عملية تطوير الوكلاء، إذ تجعلها مشابهة لهندسة البرامج العادية، ما يتيح إنشاء مهام سير عمل معقّدة للوكلاء واختبارها وتنظيمها بسهولة.

تعالج حزمة ADK محادثات المستخدمين، وتوفّر واجهة مستخدم تفاعلية على الويب، وتتصل بأدوات MCP من خلال McpToolset، وتدير الاتصالات من تطبيق إلى آخر.

بروتوكول Agent2Agent (A2A)

بروتوكول Agent2Agent (A2A) هو معيار مفتوح مصمّم لإتاحة التواصل والتعاون بين وكلاء الذكاء الاصطناعي. تمامًا كما يوحّد بروتوكول MCP الاتصالات بين الوكلاء والأدوات، يوحّد بروتوكول A2A التعاون بين الوكلاء. في عالم يتم فيه إنشاء الوكلاء عبر أُطر عمل وبيئات استضافة ومؤسسات متنوّعة، يوفّر بروتوكول A2A لغة مشتركة للاستكشاف وتفويض المهام.

تتيح حزمة ADK استخدام currency_agent كخدمة A2A (to_a2a) وتسمح لـ travel_agent باستخدامها عن بُعد كـ AgentTool (RemoteA2aAgent).

الوكلاء الفرعيون المحليون مقابل وكلاء A2A عن بُعد

تسلّط هذه البنية الضوء على نمط بالغ الأهمية في تصميم الأنظمة المتعدّدة الوكلاء:

• أداة الوكيل المحلي (weather_agent): تعمل ضمن عملية المنسّق كـ AgentTool، بدون أي وقت استجابة للشبكة أو أعباء تسلسل البروتوكول.
• أداة الوكيل عن بُعد (currency_agent): تعمل كخدمة مصغّرة مستقلة، ويتم استدعاؤها عبر الشبكة من خلال بروتوكول A2A.

أهداف الدورة التعليمية

• كيفية إنشاء خادم MCP محلي واختباره باستخدام FastMCP عبر بروتوكول HTTP القابل للبث
• كيفية إنشاء وكيل باستخدام حزمة ADK التي تستخدم أدوات MCP ‏ (McpToolset)
• كيفية عرض وكيل ADK كخادم A2A ‏ (to_a2a)
• كيفية اختبار خادم A2A باستخدام عميل Python لـ A2A ‏ (a2a.client)
• كيفية إنشاء وكيل منسِّق يجمع بين أدوات الوكلاء الفرعيين المحليين ووكلاء A2A عن بُعد
• كيفية تشغيل النظام الكامل المتعدّد الوكلاء واختباره محليًا باستخدام واجهة مستخدم الويب الخاصة بـ ADK
• كيفية وضع المكوّنات الثلاثة في حاويات ونشرها على Google Cloud Run

المتطلبات

• متصفّح، مثل Chrome
• مشروع على Google Cloud تم تفعيل الفوترة فيه

2. قبل البدء

إنشاء مشروع

إذا لم يكن لديك مشروع على Google Cloud، يمكنك إنشاء مشروع.

في Google Cloud Console، في صفحة أداة اختيار المشاريع، اختَر مشروعًا على Google Cloud أو أنشِئ مشروعًا.

تأكَّد أيضًا من تفعيل الفوترة لمشروعك على السحابة الإلكترونية. كيفية التحقّق من تفعيل الفوترة في مشروع

تفعيل Cloud Shell

‫Google Cloud Shell هي بيئة تطوير تفاعلية مستندة إلى المتصفّح، ويتم توفيرها مباشرةً ضمن Google Cloud Console. وهي أسهل طريقة لبدء استخدام Google Cloud بدون الحاجة إلى تثبيت الأدوات على الجهاز.

فعِّل Cloud Shell بالنقر على هذا الرابط. يمكنك التبديل بين Cloud Shell Terminal (لتشغيل أوامر السحابة الإلكترونية) وEditor (لعرض الملفات) من خلال النقر على الزر المقابل في Cloud Shell.

بعد الاتصال بـ Cloud Shell، تأكَّد من أنّه تمت مصادقتك:

gcloud auth list

تأكَّد من أنّ الأمر gcloud يعرف مشروعك:

gcloud config list project

اضبط رقم تعريف مشروعك:

export PROJECT_ID=<YOUR_GOOGLE_CLOUD_PROJECT_ID>
gcloud config set project $PROJECT_ID

تفعيل Cloud APIs

فعِّل واجهات برمجة التطبيقات المطلوبة لكل من Cloud Run وCloud Build وArtifact Registry وVertex AI:

gcloud services enable run.googleapis.com \
                       cloudbuild.googleapis.com \
                       artifactregistry.googleapis.com \
                       aiplatform.googleapis.com

الحصول على الشفرة‏

انسخ المستودع وانتقِل إلى دليل python:

git clone https://github.com/meteatamel/getting-started-mcp-a2a-adk.git
cd getting-started-mcp-a2a-adk/python

تثبيت التبعيات باستخدام uv

‫uv هي حزمة Python ومدير مشاريع سريع. وهي مثبّتة مسبقًا في Cloud Shell. إذا كنت تشغّل البرنامج محليًا، يمكنك تثبيته على النحو التالي:

# macOS and Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
# powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

تم تنظيم دليل Python كدليل uv موحّد

Workspace: ثبِّت جميع التبعيات في جميع أعضاء مساحة العمل (currency_mcp_server وcurrency_agent وtravel_agent) في بيئة افتراضية واحدة من خلال تشغيل:

uv sync

ضبط متغيرات البيئة

اضبط متغيّرات البيئة في ملف .env في دليل python:

الخيار "أ": Gemini Enterprise (Google Cloud / Cloud Shell)

echo "GOOGLE_GENAI_USE_ENTERPRISE=TRUE" >> .env \
&& echo "GOOGLE_CLOUD_PROJECT=$PROJECT_ID" >> .env \
&& echo "GOOGLE_CLOUD_LOCATION=global" >> .env

الخيار ب: Google AI Studio (مفتاح واجهة برمجة التطبيقات المحلي)

إذا كنت تفضّل استخدام مفتاح واجهة برمجة التطبيقات من Google AI Studio بدلاً من ذلك:

echo "GOOGLE_API_KEY=<your_api_key_here>"<>>/span>  .env \
&& echo "GOOGLE_GENAI_USE_ENTERPRISE=FALSE" >> .env

3- إنشاء خادم MCP لتحويل العملات

قبل إنشاء الوكلاء، عليك أولاً استكشاف خادم MCP الذي يعرض أداة تحويل العملات.

يتيح لك خادم MCP كتابة برامج بسيطة لعرض الإمكانات كأدوات. ويمكن بعد ذلك لوكيل (أو عدة وكلاء) الوصول إلى هذه الأدوات باستخدام بروتوكول Model Context الموحّد.

يمكن الاستفادة من حزمة Python‏ FastMCP لإنشاء خادم MCP يعرض أداة واحدة باسم get_exchange_rate. تستدعي أداة get_exchange_rate واجهة Frankfurter API للحصول على سعر الصرف الحالي بين عملتَين.

يتوفّر رمز خادم MCP في currency_mcp_server/server.py:

import asyncio
import logging
import os

import httpx
from fastmcp import FastMCP

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.dev/v1/{currency_date}",
            params={"from": currency_from, "to": currency_to},
        )
        response.raise_for_status()

        data = response.json()
        if "rates" not in data:
            logger.error(f"❌ rates not found in response: {data}")
            return {"error": "Invalid API response format."}
        logger.info(f"✅ API response: {data}")
        return data
    except httpx.HTTPError as e:
        logger.error(f"❌ API request failed: {e}")
        return {"error": f"API request failed: {e}"}
    except ValueError:
        logger.error("❌ Invalid JSON response from API")
        return {"error": "Invalid JSON response from API."}


if __name__ == "__main__":
    port = int(os.getenv("PORT", 8080))
    logger.info(f"🚀 MCP server started on port {port}")
    # 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=port,
        )
    )

بدء خادم MCP محليًا

ابدأ خادم MCP في الوحدة الطرفية:

uv run python currency_mcp_server/server.py

يبدأ الخادم على المنفذ 8080 مع توفّر نقطة نهاية HTTP قابلة للبث على /mcp (http://localhost:8080/mcp).

اختبار خادم MCP

افتح نافذة محطة طرفية ثانية (مع ترك خادم MCP قيد التشغيل في النافذة الأولى)، وانتقِل إلى مجلد python (cd getting-started-mcp-a2a-adk/python)، وشغِّل نص الاختبار البرمجي:

uv run python currency_mcp_server/test_server.py

يتصل نص الاختبار البرمجي بنقطة نهاية MCP باستخدام Client من FastMCP، ويعرض الأدوات، ويستدعي get_exchange_rate:

Connecting to MCP server at: http://localhost:8080/mcp
--- 🛠️  Tool found: get_exchange_rate ---
--- 🪛  Calling get_exchange_rate tool for USD to EUR ---
--- ✅  Success: {"amount":1.0,"base":"USD","date":"2026-09-29","rates":{"EUR":0.84919}} ---

رائع! لديك خادم MCP يعمل بشكل صحيح ويعرض أداة لتحويل العملات في الوقت الفعلي. أبقِ هذه الوحدة الطرفية مفتوحة للخطوات التالية.

4. إنشاء وكيل العملات باستخدام حزمة ADK

حان الوقت الآن لإنشاء وكيل العملات باستخدام حزمة تطوير الوكلاء (ADK).

توفّر حزمة ADK دعمًا مضمّنًا لأدوات MCP من خلال فئة McpToolset. يتصل "وكيل العملات" بخادم MCP باستخدام StreamableHTTPConnectionParams.

لنلقِ نظرة على تعريف وكيل العملات في currency_agent/agent.py:

import logging
import os
from urllib.parse import urlparse

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."
)

mcp_server_url = os.getenv("MCP_SERVER_URL", "http://localhost:8080/mcp")
if not mcp_server_url.endswith("/mcp"):
    mcp_server_url = f"{mcp_server_url.rstrip('/')}/mcp"

logger.info(f"--- 🔧 Loading MCP tools from MCP Server at {mcp_server_url}... ---")
logger.info("--- 🤖 Creating ADK Currency Agent... ---")

root_agent = LlmAgent(
    model="gemini-3.8-flash",
    name="currency_agent",
    description="An agent that can help with currency conversions",
    instruction=SYSTEM_INSTRUCTION,
    tools=[
        McpToolset(
            connection_params=StreamableHTTPConnectionParams(
                url=mcp_server_url
            )
        )
    ],
)

لاحظ كيف يكتشف McpToolset أداة get_exchange_rate ويحمّلها تلقائيًا من خادم MCP النشط عبر HTTP.

اختبار وكيل العملات باستخدام واجهة الويب الخاصة بـ ADK

يمكنك اختبار الوكيل بشكل منفصل باستخدام واجهة الويب الخاصة بالتطوير في ADK. في الوحدة الطرفية الثانية، شغِّل:

uv run adk web --allow_origins "*" currency_agent

افتح المتصفّح على http://localhost:8000 (أو انقر على معاينة الويب في Cloud Shell).

اطلب من الوكيل:

"كم يساوي 250 دولار أمريكي باليورو؟"

سيستدعي الوكيل أداة get_exchange_rate MCP ويعرض المبلغ المحوَّل. يُرجى إيقاف عملية "adk web" باستخدام "Ctrl+C" قبل المتابعة.

5- بروتوكول Agent2Agent (A2A)

بروتوكول Agent2Agent (A2A) هو معيار مفتوح مصمّم لإتاحة التواصل والتعاون بين وكلاء الذكاء الاصطناعي.

في الأنظمة الحديثة المتعدّدة الوكلاء، غالبًا ما يتم إنشاء الوكلاء باستخدام أُطر عمل مختلفة (مثل ADK وLangChain وCrewAI وAutoGen) ويتم نشرها على بنية أساسية مختلفة. تزيل واجهة برمجة التطبيقات A2A هذه العوائق من خلال تحديد عقد اتصال موحّد.

المفاهيم الأساسية لـ A2A

• اكتشاف الوكيل (بطاقة الوكيل): يعلن كل وكيل متوافق مع بروتوكول A2A عن إمكاناته وأوضاع الإدخال/الإخراج المتوافقة ومتطلبات المصادقة والمهارات من خلال ملف JSON عادي مستضاف على /.well-known/agent-card.json.
• مهارات الوكيل (AgentSkill): هي الإمكانات أو الأدوات المحدّدة التي يمكن للوكيل تنفيذها. يقرأ الوكلاء الآخرون هذه المهارات لتحديد الحالات التي يجب فيها تفويض المهام.
• التواصل عبر JSON-RPC: تتبادل الوكلاء الرسائل المنظَّمة والمهام والبيانات عبر بروتوكولات نقل عادية مثل HTTP JSON-RPC 2.0.

في الخطوة التالية، ستعرض وكيل ADK Currency Agent كخادم A2A ليتمكّن الوكلاء الآخرون من اكتشافه وتفويض مهام تحويل العملات إليه.

6. إتاحة الوصول إلى وكيل تحويل العملات كخادم A2A

تسهّل حزمة ADK تحويل أي LlmAgent إلى خادم متوافق مع A2A باستخدام الدالة الخدمية to_a2a().

راجِع الجزء السفلي من currency_agent/agent.py:

# Make the agent A2A-compatible
PORT = int(os.getenv("PORT", 8081))
AGENT_URL = os.getenv("AGENT_URL")

if AGENT_URL:
    parsed = urlparse(AGENT_URL)
    protocol = parsed.scheme or "https"
    host = parsed.hostname
    port = parsed.port or 443
else:
    protocol = "http"
    host = "localhost"
    port = PORT

a2a_app = to_a2a(root_agent, host=host, port=port, protocol=protocol)

if __name__ == "__main__":
    import uvicorn

    logger.info(f"🚀 Starting currency_agent on port {PORT}")
    uvicorn.run(a2a_app, host="0.0.0.0", port=PORT)

تنفّذ أداة المساعدة في to_a2a() الإجراءات التالية تلقائيًا:

1. يفحص أدوات LlmAgent وتعليماته.
2. يُنشئ هذا الإجراء AgentCard من تطبيق إلى تطبيق يصف مهاراته (مثل currency_agent-get_exchange_rate).
3. يتم تضمين الوكيل في تطبيق FastAPI ‏ (a2a_app) يتوافق مع نقاط نهاية JSON-RPC.
4. يتم ضبط url المُعلَن عنه ديناميكيًا في "بطاقة الوكيل" استنادًا إلى AGENT_URL (أو المضيف المحلي).

بدء خادم A2A لـ Currency Agent

في الوحدة الطرفية الثانية، ابدأ وكيل العملات:

uv run python currency_agent/agent.py

يبدأ الخادم على المنفذ 8081:

[INFO]: --- 🔧 Loading MCP tools from MCP Server at http://localhost:8080/mcp... ---
[INFO]: --- 🤖 Creating ADK Currency Agent... ---
[INFO]: 🚀 Starting currency_agent on port 8081
INFO:     Started server process
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8081 (Press CTRL+C to quit)

فحص بطاقة الوكيل

في المتصفّح، انتقِل إلى نقطة نهاية "بطاقة الوكيل": http://localhost:8081/.well-known/agent-card.json

ستظهر لك البيانات الوصفية بتنسيق JSON التي تم إنشاؤها تلقائيًا:

{
  "name": "currency_agent",
  "description": "An agent that can help with currency conversions",
  "supportedInterfaces": [
    {
      "url": "http://localhost:8081",
      "protocolBinding": "JSONRPC",
      "protocolVersion": "1.0"
    }
  ],
  "version": "0.0.1",
  "capabilities": {
    "streaming": true
  },
  "defaultInputModes": [
    "text/plain"
  ],
  "defaultOutputModes": [
    "text/plain"
  ],
  "skills": [
    {
      "id": "currency_agent",
      "name": "model",
      "description": "An agent that can help with currency conversions",
      "tags": [
        "llm"
      ]
    },
    {
      "id": "currency_agent-get_exchange_rate",
      "name": "get_exchange_rate",
      "description": "Use this to get current exchange rate.",
      "tags": [
        "llm",
        "tools"
      ]
    }
  ]
}

الاختبار باستخدام عميل A2A

افتح وحدة طرفية ثالثة، وانتقِل إلى المجلد python (cd getting-started-mcp-a2a-adk/python)، واختبِر نقطة نهاية A2A باستخدام نص برمجي مخصّص لعميل اختبار A2A:

uv run python currency_agent/test_a2aclient.py

يحلّ العميل بطاقة الوكيل باستخدام A2ACardResolver ويرسل رسالة JSON-RPC:

🔄 Fetching the agent card at http://localhost:8081
✅ Successfully fetched the agent card
====================================================
                     AgentCard                      
====================================================
Name        : currency_agent
Description : An agent that can help with currency conversions
Interfaces  : [0] http://localhost:8081  (JSONRPC 1.0)
Skills      : currency_agent, currency_agent-get_exchange_rate
====================================================
🔄 Initializing a non-streaming client
Sending request: "how much is 100 USD in GBP?"
Response:
100 USD is currently equal to approximately 73.37 GBP...

يعمل "وكيل العملات" الآن كخادم A2A يعمل بكامل طاقته.

7. إنشاء وكيل السفر

الآن، ستنشئ وكيل السفر الذي سينسّق العملية. يوضّح هذا الوكيل كيف يمكن لوكيل ADK الجمع بسهولة بين نموذجين مختلفين للأدوات:

1. أداة Local Agent Tool‏ (weather_agent): هي وكيل فرعي لحزمة ADK يعمل ضمن بيئة تشغيل Python نفسها، ما يجنّب أي تأخير في الشبكة أو التسلسل.
2. أداة الوكيل البعيد (currency_agent): وكيل خارجي متصل عبر الشبكة من خلال بروتوكول A2A باستخدام RemoteA2aAgent.

                    +-----------------------------+
                    |        travel_agent         |
                    | (Orchestrator with Web UI)  |
                    +--------------+--------------+
                                   |
             +---------------------+---------------------+
             | (In-process call)                         | (A2A Network Hop)
             v                                           v
+-------------------------+                 +-------------------------+
|   weather_agent         |                 |   currency_agent        |
|  (Local AgentTool)      |                 |  (Remote A2A Agent)     |
|  Calls wttr.in API      |                 |  Calls Currency MCP Svc |
+-------------------------+                 +-------------------------+

1. الوكيل الفرعي "الطقس المحلي"

لنلقِ نظرة على "travel_agent/subagents/weather_agent.py":

import logging
import httpx
from google.adk.agents import LlmAgent

logger = logging.getLogger(__name__)

SYSTEM_INSTRUCTION = (
    "You are a specialized assistant for providing weather forecasts and conditions for travel destinations. "
    "Your sole purpose is to use the 'get_weather' tool to fetch weather information. "
    "Provide clear, concise summaries of current weather conditions, temperatures, and forecasts to help travelers pack and plan."
)


def get_weather(location: str = "London") -> dict:
    """Gets the current weather and forecast for a given location or city."""
    logger.info(f"--- 🌤️ Tool: get_weather called for {location} ---")
    try:
        response = httpx.get(
            f"https://wttr.in/{location}?format=j1",
            headers={"User-Agent": "curl/7.68.0"},
            timeout=8.0,
        )
        response.raise_for_status()
        data = response.json()
        current = data["current_condition"][0]
        return {
            "location": location,
            "condition": current.get("weatherDesc", [{}])[0].get("value", "Unknown"),
            "temperature_celsius": f"{current.get('temp_C', 'N/A')}°C",
            "temperature_fahrenheit": f"{current.get('temp_F', 'N/A')}°F",
            "humidity": f"{current.get('humidity', 'N/A')}%",
            "wind_speed_kmph": f"{current.get('windspeedKmph', 'N/A')} km/h",
        }
    except Exception as e:
        logger.warning(f"⚠️ Weather lookup failed: {e}. Returning simulated estimate.")
        return {
            "location": location,
            "condition": "Partly Cloudy",
            "temperature_celsius": "20°C",
            "temperature_fahrenheit": "68°F",
            "humidity": "55%",
            "wind_speed_kmph": "10 km/h",
            "note": f"Live data unavailable ({e}); showing estimated conditions.",
        }


weather_agent = LlmAgent(
    model="gemini-3.8-flash",
    name="weather_agent",
    description="A local specialist agent that provides current weather reports and forecasts for travel destinations.",
    instruction=SYSTEM_INSTRUCTION,
    tools=[get_weather],
)

وكيل الطقس هو LlmAgent مزوّد بأداة دالة تجلب بيانات الطقس في الوقت الفعلي من wttr.in مع آلية احتياطية مرنة.

2. The Travel Agent Orchestrator

لنلقِ نظرة الآن على travel_agent/agent.py:

import logging
import os
from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.tools.agent_tool import AgentTool
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent, AGENT_CARD_WELL_KNOWN_PATH

try:
    from travel_agent.subagents.weather_agent import weather_agent
except ImportError:
    from subagents.weather_agent import weather_agent

logger = logging.getLogger(__name__)
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=logging.INFO)

load_dotenv()

SYSTEM_INSTRUCTION = (
    "You are a helpful travel assistant. You help users plan trips, recommend places, "
    "and answer travel-related questions.\n\n"
    "- Whenever a user asks about currency exchange rates or money conversions, "
    "delegate the request to the 'currency_agent' tool.\n"
    "- Whenever a user asks about the weather, climate, or forecasts for a destination, "
    "delegate the request to the 'weather_agent' tool."
)

CURRENCY_AGENT_URL = os.getenv("CURRENCY_AGENT_URL", "http://localhost:8081").rstrip("/")

logger.info(
    "--- 🔗 Connecting to Remote A2A Currency Agent at %s... ---",
    CURRENCY_AGENT_URL,
)

currency_remote_agent = RemoteA2aAgent(
    name="currency_agent",
    agent_card=f"{CURRENCY_AGENT_URL}{AGENT_CARD_WELL_KNOWN_PATH}",
    description="An agent that can help with currency conversions and exchange rates.",
)

logger.info("--- 🤖 Creating ADK Travel Agent... ---")

root_agent = LlmAgent(
    model="gemini-3.8-flash",
    name="travel_agent",
    description="A travel assistant that can help plan trips, check weather forecasts via the local weather agent, and convert currencies via the remote currency agent.",
    instruction=SYSTEM_INSTRUCTION,
    tools=[
        AgentTool(agent=currency_remote_agent),
        AgentTool(agent=weather_agent),
    ],
)

if __name__ == "__main__":
    import uvicorn
    from google.adk.cli.fast_api import get_fast_api_app

    PORT = int(os.getenv("PORT", 8082))
    logger.info(f"🚀 Starting travel_agent Web UI on port {PORT}")
    app = get_fast_api_app(web=True, allow_origins=["*"], agents_dir=os.path.dirname(__file__) or ".")
    uvicorn.run(app, host="0.0.0.0", port=PORT)

الميزات الرئيسية:

• يجلب "RemoteA2aAgent" بطاقة "وكيل العملات" من "http://localhost:8081/.well-known/agent-card.json".
• يتم تضمين كل من currency_remote_agent وweather_agent باستخدام AgentTool في ADK.
• يؤدي تشغيل travel_agent/agent.py إلى تشغيل واجهة مستخدم الويب الخاصة بـ ADK مباشرةً على المنفذ 8082 باستخدام get_fast_api_app(web=True).

8. تشغيل النظام المتعدّد الوكلاء واختباره على الجهاز

أصبحت الآن المكوّنات الثلاثة جاهزة. لنُشغّل النظام الكامل ونختبره على الجهاز.

تشغيل الخدمات الثلاث

تأكَّد من تشغيل الخدمات في ثلاث نوافذ منفصلة للمحطة الطرفية (من دليل getting-started-mcp-a2a-adk/python) بالترتيب التالي:

‫Terminal 1 (خادم MCP للعملات):

uv run python currency_mcp_server/server.py

(يتم تشغيله على المنفذ 8080)

‫Terminal 2 (وكيل A2A لتحويل العملات):

uv run python currency_agent/agent.py

(يتم تشغيله على المنفذ 8081)

المحطة 3 (وكيل سفر مع واجهة مستخدم على الويب):

uv run python travel_agent/agent.py

(يتم تشغيلها على المنفذ 8082)

الاختبار من خلال واجهة مستخدم ADK على الويب

افتح المتصفّح على الرابط http://localhost:8082 (أو استخدِم معاينة الويب على المنفذ 8082 في Cloud Shell).

اطرح على "وكيل السفر" سؤالاً مركّبًا يتطلّب استخدام كلتا الأداتَين:

"أخطّط لرحلة إلى لندن. ما هي حالة الطقس وكم يساوي 200 يورو بالجنيه الإسترليني؟"

شاهِد مسار التنفيذ:

1. يتلقّى وكيل السفر الطلب في واجهة مستخدم الويب.
2. يتم تفويض السؤال المتعلّق بالطقس أثناء المعالجة إلى weather_agent المحلي الذي يجلب الظروف الحالية من wttr.in.
3. ويفوّض عملية تحويل العملات عبر بروتوكول A2A إلى currency_agent على المنفذ 8081.
4. يستدعي currency_agent currency_mcp_server على المنفذ 8080 عبر MCP لجلب سعر الصرف.
5. يتم تجميع كل النتائج وتقديمها لك في إجابة واحدة مفيدة.

الاختبار من خلال سطر الأوامر

يمكنك أيضًا اختبار طلبات البحث مباشرةً من واجهة سطر الأوامر بدون فتح متصفّح:

uv run adk run travel_agent "What is the weather in London and how much is 100 USD in EUR?"

بعد أن أصبح كل شيء يعمل بسلاسة على جهازك المحلي، لننشر النظام بأكمله على Google Cloud Run.

9. نشر نظام الوكلاء المتعدّدين على Cloud Run

لكل مكوّن Dockerfile وpyproject.toml مستقلان، ما يتيح إنشاءه ونشره بشكل مستقل على Cloud Run.

باستخدام gcloud، ستنشر كل خدمة وتربط عناوين URL الخاصة بها معًا تلقائيًا.

ملاحظة مهمة: في نظام الإنتاج، لن يتم نشر خدمات Cloud Run الداخلية بدون مصادقة. يجب استخدام --no-allow-unauthenticated وفرض المصادقة باستخدام أذونات "إدارة الهوية وإمكانية الوصول" (IAM) الخاصة بـ Cloud Run Invoker (roles/run.invoker) أو هويات الخدمة أو بوابة واجهة برمجة التطبيقات. لتبسيط هذا الدرس التطبيقي حول الترميز، يتم استخدام --allow-unauthenticated حتى تتمكّن الخدمات من التواصل مع بعضها مباشرةً عبر عناوين URL المتاحة للجميع.

ضبط متغيرات البيئة

اضبط رقم تعريف مشروعك على السحابة الإلكترونية والمنطقة المستهدَفة:

export PROJECT_ID=<YOUR_GOOGLE_CLOUD_PROJECT_ID>
export REGION=us-central1

الخطوة 1: نشر خادم MCP الخاص بـ "العملات"

نشر خادم MCP مباشرةً من المصدر:

# 1. Deploy MCP server
gcloud run deploy currency-mcp-server \
  --source currency_mcp_server \
  --region $REGION \
  --allow-unauthenticated

# 2. Capture the deployed MCP server URL
MCP_SERVER_URL=$(gcloud run services describe currency-mcp-server --region $REGION --format='value(status.url)')/mcp
echo "MCP Server URL: $MCP_SERVER_URL"

الخطوة 2: نشر "وكيل العملات"

بما أنّه يتم تخصيص عنوان URL العلني لـ "currency-agent" عند نشره لأول مرة، يجب نشر الخدمة أولاً، ثم الحصول على عنوان URL الخاص بها، وتعديل متغيّر البيئة AGENT_URL الخاص بها لكي تعلن بطاقة وكيل A2A عن عنوان HTTPS العلني الصحيح:

# 1. Deploy Currency Agent connected to the deployed MCP Server
gcloud run deploy currency-agent \
  --source currency_agent \
  --region $REGION \
  --allow-unauthenticated \
  --update-env-vars MCP_SERVER_URL="$MCP_SERVER_URL",GOOGLE_GENAI_USE_ENTERPRISE="true",GOOGLE_CLOUD_PROJECT="$PROJECT_ID",GOOGLE_CLOUD_LOCATION="global"

# 2. Capture its assigned Cloud Run URL
CURRENCY_AGENT_URL=$(gcloud run services describe currency-agent --region $REGION --format='value(status.url)')
echo "Currency Agent URL: $CURRENCY_AGENT_URL"

# 3. Update AGENT_URL so the agent advertises its public HTTPS endpoint in its Agent Card
gcloud run services update currency-agent \
  --region $REGION \
  --update-env-vars AGENT_URL="$CURRENCY_AGENT_URL"

الخطوة 3: نشر وكيل السفر

الآن، انشروا "وكيل السفر" واضبطوه باستخدام CURRENCY_AGENT_URL:

# 1. Deploy Travel Agent (with ADK Web UI) connected to Currency Agent
gcloud run deploy travel-agent \
  --source travel_agent \
  --region $REGION \
  --allow-unauthenticated \
  --update-env-vars CURRENCY_AGENT_URL="$CURRENCY_AGENT_URL",GOOGLE_GENAI_USE_ENTERPRISE="true",GOOGLE_CLOUD_PROJECT="$PROJECT_ID",GOOGLE_CLOUD_LOCATION="global"

# 2. Capture its assigned Cloud Run URL
TRAVEL_AGENT_URL=$(gcloud run services describe travel-agent --region $REGION --format='value(status.url)')
echo "Travel Agent URL: $TRAVEL_AGENT_URL"

ملاحظة: يمكنك أيضًا نشر travel_agent باستخدام أمر نشر Cloud Run المضمّن في ADK مع علامة --with_ui:

uv run adk deploy cloud_run travel_agent \
  --project $PROJECT_ID \
  --region $REGION \
  --with_ui \
  --env CURRENCY_AGENT_URL="$CURRENCY_AGENT_URL"

10. اختبار نظام الوكلاء المتعددين في Cloud Run

بعد نشر الخدمات المصغّرة الثلاث، لنتحقّق من كل طبقة في Cloud Run.

1. اختبار خادم MCP البعيد

اختبِر خادم MCP المنشور على Cloud Run من سطر الأوامر:

MCP_SERVER_URL="$MCP_SERVER_URL" uv run python currency_mcp_server/test_server.py

من المفترض أن تظهر الأداة مدرَجة وتم استدعاؤها بنجاح على Cloud Run.

2. اختبار وكيل العملات عبر A2A

اختبِر وكيل العملات الذي تم نشره باستخدام برنامج اختبار A2A:

AGENT_URL="$CURRENCY_AGENT_URL" uv run python currency_agent/test_a2aclient.py

سيجلب العميل "بطاقة الوكيل" عن بُعد من Cloud Run ويتبادل الرسائل عبر HTTPS JSON-RPC.

3- اختبار واجهة المستخدم على الويب لوكيل السفر في Cloud Run

افتح "$TRAVEL_AGENT_URL" في متصفّحك.

يتم عرض واجهة مستخدم الويب الكاملة لـ ADK مباشرةً من Cloud Run. جرّب طرح سؤال:

"أخطّط لرحلة إلى طوكيو. ما هي حالة الطقس؟ وكم يساوي 500 دولار أمريكي بالين الياباني؟"

[Browser] 
   │
   ▼ (HTTPS)
[Cloud Run: travel-agent]
   ├──> Local in-process tool: weather_agent (wttr.in)
   └──> A2A JSON-RPC (HTTPS)
           │
           ▼
        [Cloud Run: currency-agent]
           └──> MCP Streamable HTTP (HTTPS)
                   │
                   ▼
                [Cloud Run: currency-mcp-server] (Frankfurter API)

تتوسع الخدمات الثلاث بشكل مستقل على بنية أساسية بدون خادم، وتتواصل بسلاسة عبر بروتوكولات مفتوحة.

11. تهانينا وتنظيف

إنجازاتك

تهانينا! لقد أنشأت نظامًا متعدد الوكلاء على مستوى الإنتاج ونشرته:

• بروتوكول سياق النموذج (MCP): تم إنشاء خادم FastMCP يعرض أدوات واجهة برمجة التطبيقات المباشرة عبر بروتوكول HTTP القابل للبث.
• حزمة تطوير الوكلاء (ADK): تم إنشاء وكلاء معياريين باستخدام LlmAgent، وربط أدوات MCP بـ McpToolset، واستضافة واجهة مستخدم الويب الخاصة بحزمة ADK.
• بروتوكول Agent2Agent (A2A): تم عرض وكيل كخدمة A2A (to_a2a)، ونشر بطاقات الوكلاء (agent-card.json)، واستخدام وكلاء عن بُعد من خلال RemoteA2aAgent.
• بنية الوكلاء المتعددين: تم دمج الوكلاء الفرعيين المحليين ضمن العملية (weather_agent) مع الوكلاء البعيدين الموزّعين (currency_agent).
• النشر بدون خادم في Cloud Run: تم وضع الخدمات الثلاث في حاويات ونشرها باستخدام إعدادات ربط مبرمَجة.

للاطّلاع على الرمز الكامل (بما في ذلك عمليات تنفيذ Python وJava) والتحديثات المستقبلية، انتقِل إلى مستودع getting-started-mcp-a2a-adk.

تَنظيم

لتجنُّب تحمُّل رسوم على حسابك على Google Cloud مقابل خدمات Cloud Run التي تم إنشاؤها في هذا الدرس التطبيقي حول الترميز، اتّبِع الخطوات التالية:

احذف خدمات Cloud Run المنشورة:

gcloud run services delete currency-mcp-server --region $REGION --quiet
gcloud run services delete currency-agent --region $REGION --quiet
gcloud run services delete travel-agent --region $REGION --quiet

بدلاً من ذلك، إذا أنشأت مشروعًا مخصّصًا لهذا الدرس التطبيقي حول الترميز، يمكنك إيقاف المشروع بأكمله:

gcloud projects delete $PROJECT_ID