1. نظرة عامة
في هذا الدرس التطبيقي حول الترميز، ستنشئ نظامًا متعدد الوكلاء تتواصل فيه عدّة وكلاء من "حزمة تطوير الوكلاء" (ADK) ويتعاونون باستخدام بروتوكول Agent2Agent (A2A).
أهداف الدورة التعليمية
- كيفية إنشاء وكلاء ADK مستقلين متعددين
- كيفية منح كل وكيل بطاقة وكيل وتغليفها كخوادم A2A
- كيفية إنشاء وكيل مضيف ينسّق الوكلاء البعيدين
- كيفية إنشاء اتصالات الوكيل البعيد
- كيفية اختبار نظام الوكلاء المتعدّدين محليًا
المتطلبات
- مشروع Google Cloud تم تفعيل الفوترة فيه
- متصفّح ويب، مثل Chrome
- Python 3.12 والإصدارات الأحدث وuv، إذا كنت ستشغّل الرمز البرمجي على جهازك بدلاً من Cloud Shell
هذا الدرس التطبيقي حول الترميز مخصّص للمطوّرين المتوسطي الخبرة الذين لديهم بعض المعرفة بلغة Python وGoogle Cloud.
يستغرق إكمال هذا الدرس التطبيقي حول الترميز حوالي 15 دقيقة.
يجب أن تكون تكلفة الموارد التي تم إنشاؤها في هذا الدرس التطبيقي حول الترميز أقل من 5 دولارات أمريكية.
2. إعداد البيئة
إنشاء مشروع على Google Cloud
- في Google Cloud Console، في صفحة اختيار المشروع، اختَر مشروعًا على السحابة الإلكترونية أو أنشِئ مشروعًا على السحابة الإلكترونية.
- تأكَّد من تفعيل الفوترة لمشروعك على السحابة الإلكترونية. كيفية التحقّق مما إذا كانت الفوترة مفعَّلة في مشروع
بدء Cloud Shell Editor
لبدء جلسة Cloud Shell من "وحدة تحكّم Google Cloud"، انقر على تفعيل Cloud Shell في "وحدة تحكّم Google Cloud".
سيؤدي ذلك إلى بدء جلسة في اللوحة السفلية من "وحدة تحكّم Google Cloud".
لتشغيل المحرِّر، انقر على فتح المحرِّر في شريط الأدوات في نافذة Cloud Shell.
تفعيل واجهات برمجة التطبيقات
في الوحدة الطرفية، نفِّذ الأمر التالي لتفعيل واجهة برمجة التطبيقات التي يستخدمها "حزمة تطوير التطبيقات" لطلب بيانات من Gemini:
gcloud services enable aiplatform.googleapis.com
ضبط إعدادات البيئة
ابدأ بتنفيذ الأمر التالي في الوحدة الطرفية لإنشاء بنية مجلد المشروع لنظام A2A. لأغراض هذا العرض التوضيحي، سنستخدم مسارًا مطلقًا من دليل $HOME:
mkdir -p $HOME/app/src/agents $HOME/app/src/host
touch $HOME/app/.env $HOME/app/pyproject.toml
بعد أن أصبح لدينا التصميم العام، لنملأ إعدادات البيئة. انسخ مقطع الرمز التالي في ملف .env الجديد:
املأ ملف .env الجديد برقم تعريف مشروعك على Google Cloud. يجب إبقاء GOOGLE_CLOUD_LOCATION مضبوطًا على global، حيث يتوفّر نموذج Gemini في هذا الدرس العملي. يمكنك أيضًا إدخال عنوان بريد إلكتروني من اختيارك في MAIL_TO إذا كنت تريد تلقّي رسائل إلكترونية تتضمّن التقارير من الوكيل الذي ستنشئه. يمكنك بعد ذلك إضافة رمز مميّز للوصول الشخصي إلى GitHub GITHUB_TOKEN لتسهيل عمل وكيل استرداد البيانات من GitHub.
افتح ملف .env الجديد باستخدام أمر bash التالي:
cloudshell edit $HOME/app/.env
ثم انسخ الملف التالي في ملف .env في app/.env. ملاحظة مهمة: تأكَّد من إدخال قيمك الخاصة.
# --- Google Cloud ---
GOOGLE_GENAI_USE_ENTERPRISE=True
GOOGLE_CLOUD_PROJECT=<your-gcp-project-id>
GOOGLE_CLOUD_LOCATION=global
# --- GitHub ---
# Personal Access Token with "repo" scope
# Create one at: https://github.com/settings/tokens
# Generate new token --> Fine-grained, repo-secured --> (populate) Token name --> (scroll down) Generate token
GITHUB_TOKEN=<your-github-pat>
GITHUB_MCP_URL=https://api.githubcopilot.com/mcp/
TARGET_REPOS=["google/adk-python", "langchain-ai/langchain"]
DEFAULT_ISSUE_COUNT=5
DEFAULT_PR_COUNT=5
# --- Email (Optional) ---
# Only needed if you want the emailer agent to send reports
RESEND_API_KEY=<your-resend-API-key>
RESEND_DOMAIN=<your-domain>
MAIL_TO=<insert-email-to-receive-reports>
# --- Local Development Overrides ---
RETRIEVAL_AGENT_URL=http://localhost:8001
EVALUATOR_AGENT_URL=http://localhost:8002
EMAILER_AGENT_URL=http://localhost:8003
ORCHESTRATOR_PORT=http://localhost:8000
بعد ملء متغيرات البيئة، علينا ضبط بيئة uv. افتح ملف pyproject.toml:
cloudshell edit $HOME/app/pyproject.toml
انسخ مقطع الرمز التالي في ملف pyproject.toml:
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "app"
version = "0.1.0"
description = "Multi-agent system designed to assess the newest issues and pull requests from numerous agentic development platforms"
authors = [
{ name = "Thomas Wagner" },
{ name = "Kris Overholt" }
]
license = "Apache-2.0"
requires-python = ">=3.11"
dependencies = [
"markdown",
"resend",
"uvicorn",
"a2a-sdk",
"google-adk[a2a,mcp]",
"httpx",
"python-dotenv",
]
[project.optional-dependencies]
test = [
"pytest",
"pytest-asyncio",
"pytest-mock",
"respx",
]
[tool.ruff]
target-version = "py313"
line-length = 80
[tool.ruff.lint]
select = ["E", "F", "I", "C", "PL", "B", "UP", "RUF"]
ignore = ["E501", "C901"]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]
[tool.ruff.lint.isort]
known-first-party = ["src"]
[tool.hatch.build.targets.wheel]
packages=["src/"]
تثبيت الحِزم التابعة
الآن، شغِّل الأوامر التالية في الوحدة الطرفية للانتقال إلى دليل app الذي أنشأته للتو وتثبيت جميع الاعتمادات اللازمة من ملف pyproject.toml:
cd $HOME/app
uv sync
تنشئ uv بيئة Python معزولة لهذا الدرس التطبيقي حول الترميز، لذا لن تحتاج إلى تفعيل أي شيء. ابدأ أوامر Python بالرمز uv run.
أصبح لدينا الآن كل ما نحتاجه لإنشاء نظام متعدد الوكلاء.
3- إنشاء وكيل Retriever
إيفا هي مطوّرة برامج تريد البقاء على اطّلاع على مستودعات GitHub عند تعديلها بإضافة مشاكل جديدة وطلبات سحب. لذا، أنشأت وكيل ADK لاسترداد بيانات GitHub التي طلبتها.
لأغراض هذا العرض التوضيحي، نفِّذ الأمر التالي من جذر مشروعك لإنشاء الدليل والملف اللازمَين لوكيل الاسترجاع:
mkdir -p $HOME/app/src/agents/retriever
touch $HOME/app/src/agents/retriever/agent.py
touch $HOME/app/src/agents/retriever/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/retriever/__init__.py
انسخ مقطع رمز ADK التالي إلى $HOME/app/src/agents/retriever/agent.py:
import logging
import os
import json
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
from google.adk.tools.mcp_tool.mcp_session_manager import (
StreamableHTTPConnectionParams,
)
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
logger = logging.getLogger(__name__)
load_dotenv()
GITHUB_MCP_URL = os.getenv(
"GITHUB_MCP_URL", "https://api.githubcopilot.com/mcp/"
)
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
if GITHUB_TOKEN is None:
raise ValueError("GITHUB_TOKEN env is not set")
TARGET_REPOS = os.getenv("TARGET_REPOS")
DEFAULT_ISSUE_COUNT = os.getenv("DEFAULT_ISSUE_COUNT")
DEFAULT_PR_COUNT = os.getenv("DEFAULT_PR_COUNT")
# --- Prompt ---
GITHUB_RETRIEVAL_INSTRUCTIONS = f"""
You are a specialized GitHub data retrieval agent.
Your only purpose is to fetch data from GitHub using the available MCP tools and format it for another agent.
**DEFAULT CONFIGURATION:**
If the orchestrator does not specify which repositories or how many items to fetch, you MUST use the following defaults:
- **Repositories:** {TARGET_REPOS}
- **PR Count:** {DEFAULT_PR_COUNT} per repository
- **Issue Count:** {DEFAULT_ISSUE_COUNT} per repository
**Your Task:**
1. Analyze the task. If no specific repo is mentioned, iterate through the Default Repositories list above.
2. Use the `GitHub` MCP tool to fetch the requested data (Pull Requests and/or Issues).
3. **CRITICAL:** After the tool has finished running, you MUST take the raw output and compile it into a single response.
The orchestrator is waiting for this raw data to pass to the Evaluator. Do not summarize it.
"""
# --- Agent Initialization ---
root_agent = Agent(
model=Gemini(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
name="retriever",
instruction=GITHUB_RETRIEVAL_INSTRUCTIONS,
description="""
Connects to the GitHub MCP server to retrieve real-time development
insights, actively reading repository issues, pull requests, and commit
histories.
""",
tools=[
McpToolset(
connection_params=StreamableHTTPConnectionParams(
url=GITHUB_MCP_URL,
headers={
"Authorization": f"Bearer {GITHUB_TOKEN}",
"X-MCP-Toolsets": "repos,issues,pull_requests",
"X-MCP-Readonly": "true",
},
)
)
],
)
يعمل هذا الوكيل كأداة مستقلة. لاختبارها بشكل مستقل، شغِّل الأوامر التالية للدردشة معها باستخدام adk run من الدليل agents:
cd $HOME/app/src/agents
uv run adk run retriever
اكتب رسالة في طلب [user]: لتجربتها، ثم اكتب exit لإيقاف الوكيل.
4. إنشاء وكيل التقييم
يجد "ريتشارد" في كثير من الأحيان أنّه عليه تبسيط الكثير من المصطلحات الفنية لعملائه وزملائه غير المتخصصين في هذا المجال. وبعد أن سئم "ريتشارد" من الحاجة إلى تعريف المصطلحات نفسها وشرح المشروع نفسه بطريقة موجزة، أنشأ وكيل تقطير يلخّص المصطلحات الفنية في نص سهل الفهم.
لأغراض هذا العرض التوضيحي، شغِّل الأمر التالي لإنشاء الملف الخاص بعميل التقييم:
mkdir -p $HOME/app/src/agents/evaluator
touch $HOME/app/src/agents/evaluator/agent.py
touch $HOME/app/src/agents/evaluator/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/evaluator/__init__.py
انسخ مقطع رمز الوكيل أدناه والصِقه في $HOME/app/src/agents/evaluator/agent.py
import json
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
load_dotenv()
# --- Prompt ---
EVAL_AND_SUMMARIZATION_INSTRUCTIONS = """
You are a specialized analysis and summarization agent. Your only purpose is to take raw, structured text about GitHub repositories and transform it into a concise, human-readable Markdown report.
**Your Task:**
1. You will receive a block of text containing raw data about pull requests and issues from an orchestrator.
2. Analyze the provided data. For pull requests, evaluate their significance. For issues, identify key themes and problems.
3. **CRITICAL:** You MUST generate a comprehensive summary in Markdown format based on your analysis. Your final output should be ONLY this Markdown report. Do not add any conversational text or explanations (e.g., "Here is the summary..."). The orchestrator needs to pass your clean report to another agent or directly to the user.
**Output Format Rules:**
- The report MUST be in Markdown.
- Structure the report by repository.
- For each repository, provide a concise overview of significant pull requests and important issues.
- Conclude with overall insights.
The orchestrator is waiting for this report. Ensure your final response consists of nothing but the complete Markdown summary.
"""
# --- Agent ---
root_agent = Agent(
model=Gemini(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
name="evaluator",
instruction=EVAL_AND_SUMMARIZATION_INSTRUCTIONS,
description="""
Distills and summarizes complex GitHub data, breaking down pull
requests, code changes, and issue discussions into easily understandable
insights.
""",
)
يعمل هذا الوكيل كأداة مستقلة. لاختبارها بشكل مستقل، شغِّل الأوامر التالية في وحدة طرفية جديدة للدردشة معها باستخدام adk run من الدليل agents:
cd $HOME/app/src/agents
uv run adk run evaluator
اكتب رسالة في طلب [user]: لتجربتها، ثم اكتب exit لإيقاف الوكيل.
5- إنشاء وكيل إرسال الرسائل الإلكترونية
سئم إيفان من عدد الرسائل الإلكترونية التي عليه كتابتها والتي تتضمّن تلخيص نص متوفّر بسهولة وإعادة تنسيقه. لذلك، أنشأ وكيلًا لإرسال الرسائل الإلكترونية يعيد تنسيق نص معيّن ويرسله إلى حساب بريد إلكتروني محدّد.
لأغراض هذا العرض التوضيحي، نفِّذ الأمر التالي في الوحدة الطرفية لإنشاء ملف لوكيل إرسال الرسائل الإلكترونية:
mkdir -p $HOME/app/src/agents/emailer
touch $HOME/app/src/agents/emailer/agent.py
touch $HOME/app/src/agents/emailer/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/emailer/__init__.py
انسخ مقطع رمز الوكيل التالي إلى app/src/agents/emailer/agent.py
import logging
import os
import markdown
import resend
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
load_dotenv()
logger = logging.getLogger(__name__)
RESEND_API_KEY = os.getenv("RESEND_API_KEY")
RESEND_DOMAIN = os.getenv("RESEND_DOMAIN")
MAIL_TO = os.getenv("MAIL_TO")
if RESEND_API_KEY:
resend.api_key = RESEND_API_KEY
# --- Tools ---
def send_report_email(recipient: str, subject: str, body: str) -> str:
"""
Sends an email of the given subject and body to the specified recipient
via Resend. If recipient is None, it defaults to the MAIL_TO environment variable.
Args:
recipient (str | None): The email address of the recipient. If None, defaults to MAIL_TO.
subject (str): The subject of the email.
body (str): The body of the email.
Returns:
str: A success or failure message.
"""
if not recipient:
recipient = MAIL_TO
if not all([RESEND_API_KEY, RESEND_DOMAIN, recipient]):
error_msg = "Error: Email tool configuration missing (API Key, Domain, or Recipient)"
logger.error(error_msg)
return error_msg
try:
html_body = markdown.markdown(body, extensions=["tables"])
params: resend.Emails.SendParams = {
"from": f"Research Agent <agent@{RESEND_DOMAIN}>",
"to": [recipient], #type: ignore
"subject": subject,
"text": body,
"html": html_body,
}
email = resend.Emails.send(params)
logger.info(f"Email sent successfully. ID: {email['id']}")
return f"Email sent! ID: {email['id']}"
except Exception as e:
error_msg = f"Failed to send email: {e}"
logger.error(error_msg)
return error_msg
# --- Prompt ---
EMAIL_INSTRUCTIONS = """
You are an emailer agent responsible for formatting and sending a research report via email.
INPUT: You will receive a Markdown-formatted string (the report) and the email recipient.
YOUR TASK:
1. Take the Markdown content and format it appropriately for an email body.
The goal is to render the Markdown effectively so it is readable and well-presented in an email client.
2. Based on the summary, generate a concise and informative subject line for the email.
The subject line should reflect the main themes or key insights from the report.
3. Use the `send_report_email` tool with the extracted recipient, the generated subject line, and the formatted email body.
If no email recipient is provided, output a message indicating that no recipient was specified and do NOT call the tool.
"""
# --- Agent ---
root_agent = Agent(
model=Gemini(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
name="emailer",
instruction=EMAIL_INSTRUCTIONS,
description="""
Acts as the final delivery mechanism in the pipeline, dispatching
generated summaries and reports to designated email addresses.
""",
tools=[
send_report_email
],
)
يعمل هذا الوكيل كأداة مستقلة. لاختبارها بشكل مستقل، شغِّل الأوامر التالية في وحدة طرفية جديدة للدردشة معها باستخدام adk run من الدليل agents:
cd $HOME/app/src/agents
uv run adk run emailer
اكتب رسالة في طلب [user]: لتجربتها، ثم اكتب exit لإيقاف الوكيل.
6. إنشاء بطاقات الوكلاء
لقد استعرضنا للتو ثلاث قصص لمطوّرين مستقلين لكل منهم وكيل فريد. يتم نشر جميع هؤلاء الوكلاء وتسجيل الدخول إليهم في "مكتبة الوكلاء" الخاصة بشركتهم. تريد مطوّرة أخرى، هي ديان، جمع هذه الأفكار الفردية في نظام واحد متعدد الوكلاء.
الهدف الذي تريد تحقيقه هو الحصول على وكيل بحث عند الطلب يبقيها على اطّلاع دائم على المشاكل وطلبات الدمج من أُطر تطوير الوكلاء المختلفة. وتريد أيضًا أن يرسل ملخّصًا موجزًا عبر البريد الإلكتروني عن جميع التطورات الجديدة في النظام المتكامل. بما أنّ جميع البرامج يتم تطويرها بشكل فردي في بيئات مختلفة، فإنّ A2A هو الحلّ المثالي لدمج هذه البرامج الحالية.
تتمثّل الخطوة الأولى في تجميع سلسلة من الوكلاء البعيدين في منح كل وكيل بعيد مطلوب "بطاقة وكيل". يمكن اعتبارها بطاقات أعمال خاصة بالوكيل. وهي ملفات JSON تُستخدَم لمنح الوكيل المضيف القدرة على تحديد الوكلاء من خلال أسمائهم وأوصافهم وإمكاناتهم ومخططات الإدخال والإخراج. تستخدِم البطاقات في هذا الدرس التطبيقي حول الترميز تنسيق "بطاقة الوكيل" A2A الإصدار 1.0، والذي يسرد كل نقطة نهاية يمكن الوصول إلى الوكيل فيها في supportedInterfaces.
أولاً، نفِّذ الأمر التالي من جذر مشروعك لإنشاء الدليل cards وجميع ملفات JSON اللازمة لبطاقات الوكيل:
mkdir -p $HOME/app/cards/
touch $HOME/app/cards/github_retrieval_agent_card.json
touch $HOME/app/cards/content_evaluator_agent_card.json
touch $HOME/app/cards/emailer_agent_card.json
لا يمكن الوصول إلى هذه الملفات إلا من خلال أمر bash التالي:
cloudshell edit $HOME/app/cards/github_retrieval_agent_card.json
انسخ محتوى JSON التالي في الملف app/cards/github_retrieval_agent_card.json:
{
"name": "GitHub_Retrieval_Agent",
"description": "Connects to the GitHub MCP server to retrieve real-time development insights, actively reading repository issues, pull requests, and commit histories.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8001",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"json",
"text/plain"
],
"skills": [
{
"id": "read_github_repos",
"name": "GitHub_Retriever",
"description": "Fetches and analyzes recent content updates from GitHub repositories, including open/closed issues, pull request statuses, and detailed commit logs.",
"tags": [
"Find the newest pull requests from",
"Check recent issues in",
"Summarize commits for",
"GitHub repository status"
],
"examples": [
"Find the 10 most recent pull requests from the langchain-ai/langgraph repository along with their commits.",
"List all open issues tagged with 'bug' in the current repository."
]
}
]
}
انسخ المحتوى التالي في ملف app/cards/content_evaluator_agent_card.json. سيقدّم هذا الوصف للوكيل المضيف معلومات عن الإجراءات التي يمكن لهذا الوكيل تنفيذها وأنواع البيانات التي يمكنك تقديمها له والنتائج التي سيعرضها.
كما في السابق، استخدِم هذا الأمر لتعديل بطاقة وكيل التقييم:
cloudshell edit $HOME/app/cards/content_evaluator_agent_card.json
{
"name": "Content_Evaluation_Agent",
"description": "Distills and summarizes complex GitHub data, breaking down pull requests, code changes, and issue discussions into easily understandable insights.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8002",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"defaultInputModes": [
"text",
"json",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"skills": [
{
"id": "evaluate_github_content",
"name": "Evaluate GitHub Content",
"description": "Analyzes and synthesizes provided GitHub data—including code diffs, commit histories, issue threads, and PR comments—into clear, structured summaries.",
"tags": [
"summarize pull request",
"evaluate GitHub changes",
"break down commits",
"distill issue discussion",
"code review summary"
],
"examples": [
"Make a thorough summary of the code changes and commit history from this pull request data.",
"Break down the main arguments and proposed solutions from this provided GitHub issue discussion."
]
}
]
}
في ما يلي بطاقة الوكيل الخاصة بوكيل إرسال الرسائل الإلكترونية. انسخها إلى الملف app/cards/emailer_agent_card.json باستخدام أمر cloudshell نفسه:
cloudshell edit $HOME/app/cards/emailer_agent_card.json
{
"name": "Email_Agent",
"description": "Acts as the final delivery mechanism in the pipeline, dispatching generated summaries and reports to designated email addresses.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8003",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"skills": [
{
"id": "dispatch_email_report",
"name": "Dispatch_Report_via_Email",
"description": "Takes synthesized text data and securely emails it to a specified recipient or distribution list.",
"tags": [
"send email",
"email summary",
"dispatch report",
"forward to inbox"
],
"examples": [
"Email me the summarized evaluations from the retrieved pull request and issue data.",
"Take this code review breakdown and send it in an email to the dev team."
]
}
]
}
7. إنشاء منفّذات الوكيل البعيد
لكي يتمكّن الوكيل المضيف من "استدعاء" الوكلاء البعيدين أو "التحدّث معهم"، يجب أن يكون لكل وكيل برنامج تنفيذي ظاهر لخادم A2A، ويتم تحديده من خلال "بطاقة الوكيل". المُنفِّذ هو فئة مجرّدة تتضمّن طريقتَي execute() وcancel()، وستستدعي الوكيل البعيد وتزوّده بمهمة أو رسالة، أو ستلغي المهمة تمامًا.
في ما يلي تنفيذ لبرنامج تنفيذي مجرّد لوكيل ADK يسهّل إرسال الرسائل وترتيب المهام للوكلاء البعيدين، وكل ذلك مع تبادل عناصر Part النصية. لهذا السبب، يمكننا استخدام أداة تنفيذ مشتركة للوكلاء بين الوكلاء. أنشئ ملفًا جديدًا لتنفيذ الأوامر:
touch $HOME/app/src/agents/executor.py
انسخ مقطع الرمز التالي في ملف $HOME/app/src/agents/executor.py الجديد:
# $HOME/app/src/agents/executor.py
from a2a.helpers import new_task_from_user_message, new_text_message
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import Part, TaskState
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
class BaseAgentExecutor(AgentExecutor):
"""An AgentExecutor that runs a remote ADK Agent"""
def __init__(
self,
agent,
status_message='Processing request...',
artifact_name='response',
):
"""Initialize a generic ADK agent executor.
Args:
agent: The ADK agent instance
status_message: Message to display while processing
artifact_name: Name for the response artifact
"""
self.agent = agent
self.status_message = status_message
self.artifact_name = artifact_name
self.runner = Runner(
app_name=agent.name,
agent=agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
task = context.current_task or new_task_from_user_message(context.message) # type: ignore
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, task.context_id)
user_id = context.call_context.user.user_name or "a2a_user"
try:
# Update status with custom message
await updater.update_status(
TaskState.TASK_STATE_WORKING,
new_text_message(
self.status_message,
context_id=task.context_id,
task_id=task.id,
),
)
# Process with ADK agent, reusing the session for this context
session = await self.runner.session_service.get_session(
app_name=self.agent.name,
user_id=user_id,
session_id=task.context_id,
)
if session is None:
session = await self.runner.session_service.create_session(
app_name=self.agent.name,
user_id=user_id,
state={},
session_id=task.context_id,
)
content = types.Content(
role="user", parts=[types.Part.from_text(text=query)]
)
response_text = ""
async for event in self.runner.run_async(
user_id=user_id, session_id=session.id, new_message=content
):
if event.is_final_response() and event.content and event.content.parts:
for part in event.content.parts:
if hasattr(part, "text") and part.text:
response_text += part.text + "\n"
elif hasattr(part, "function_call"):
# Log or handle function calls if needed
pass # Function calls are handled internally by ADK
# Add response as artifact with custom name
await updater.add_artifact(
[Part(text=response_text)],
name=self.artifact_name,
)
await updater.complete()
except Exception as e:
await updater.update_status(
TaskState.TASK_STATE_FAILED,
new_text_message(
f"Error: {e!s}",
context_id=task.context_id,
task_id=task.id,
),
)
async def cancel(
self,
context: RequestContext,
event_queue: EventQueue
) -> None:
"""Cancel the execution of a specific task."""
raise NotImplementedError("Cancel not implemented for RetrieverAgentExecutor")
بعد أن أصبح لدينا منفّذ وكيل مجرّد، يمكننا استخدام ميزة الوراثة في لغة Python لضبط المنفّذ بدقة بما يتناسب مع إمكانات واحتياجات كل وكيل. في حالتنا، يكون تنفيذ استدعاء الوكيل باستخدام حمولة، وانتظار الرد، واسترداد حمولة الرد موحدًا لسير العمل. لهذا السبب، لا نحتاج إلى إجراء أي تعديلات معيّنة. ومع ذلك، يحتاج كل خادم من خوادم A2A إلى برنامج تنفيذي للوكيل. نفِّذ الأمر التالي لإنشاء ملفات التنفيذ لجميع الوكلاء الثلاثة:
touch $HOME/app/src/agents/retriever/executor.py
touch $HOME/app/src/agents/evaluator/executor.py
touch $HOME/app/src/agents/emailer/executor.py
الآن، أضِف المحتوى المناسب إلى كل ملف.
# $HOME/app/src/agents/retriever/executor.py
from ..executor import BaseAgentExecutor
class RetrieverAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Retriever Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
# $HOME/app/src/agents/evaluator/executor.py
from ..executor import BaseAgentExecutor
class EvaluatorAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Evaluator Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
# $HOME/app/src/agents/emailer/executor.py
from ..executor import BaseAgentExecutor
class EmailerAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Emailer Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
8. السماح لخوادم A2A بالوصول إلى الوكلاء البعيدين
بعد أن أصبح لكل وكيل بطاقة عمل خاصة به، يمكن اكتشافه عند عرضه على نقطة نهاية من خلال خادم A2A. لأغراض هذا الدرس التطبيقي حول الترميز، سنبقي العملية بأكملها محلية باستخدام المضيفات المحلية لكل نقطة من نقاط نهاية الوكيل البعيد. ولكن في الواقع، يمكن عرضها على أي نقطة نهاية مطلوبة، ويمكن اكتشافها طالما تمت الإشارة إليها في url الحقل supportedInterfaces في بطاقات الوكيل.
يعرض كل خادم بطاقة الوكيل الخاصة به على /.well-known/agent-card.json، وهو المكان الذي يبحث فيه عملاء A2A عن البطاقة، ويتعامل مع طلبات A2A JSON-RPC على /.
الخطوة التالية لكل وكيل بعيد هي إتاحة الوصول إليه من خلال خادم A2A الخاص به. نفِّذ الأمر التالي لإنشاء ملف server.py لجميع الوكلاء البعيدين الثلاثة:
touch $HOME/app/src/agents/retriever/server.py
touch $HOME/app/src/agents/evaluator/server.py
touch $HOME/app/src/agents/emailer/server.py
الآن، ستضيف رمز الخادم لكلّ وكيل. بدءًا بوكيل الاسترجاع:
# $HOME/app/src/agents/retriever/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as retriever_agent
from .executor import RetrieverAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/github_retrieval_agent_card.json", "r") as f:
card_data = json.load(f)
github_retrieval_agent_card = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=RetrieverAgentExecutor(
agent=retriever_agent
),
task_store=InMemoryTaskStore(),
agent_card=github_retrieval_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(github_retrieval_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8001))
print(f"Starting Retriever Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
بعد ذلك، أنشئ الخادم الخاص بوكيل التقييم:
# $HOME/app/src/agents/evaluator/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as evaluator_agent
from .executor import EvaluatorAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/content_evaluator_agent_card.json", "r") as f:
card_data = json.load(f)
content_evaluator_agent_card = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=EvaluatorAgentExecutor(agent=evaluator_agent),
task_store=InMemoryTaskStore(),
agent_card=content_evaluator_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(content_evaluator_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8002))
print(f"Starting Evaluator Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
وأخيرًا، بالنسبة إلى وكيل إرسال الرسائل الإلكترونية:
# $HOME/app/src/agents/emailer/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as emailer_agent
from .executor import EmailerAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/emailer_agent_card.json", "r") as f:
card_data = json.load(f)
emailer_agent_card = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=EmailerAgentExecutor(
agent=emailer_agent
),
task_store=InMemoryTaskStore(),
agent_card=emailer_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(emailer_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8003))
print(f"Starting Emailer Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
9. إنشاء وكيل المضيف
ما فعلناه حتى الآن هو تحويل كل وكيل من وكلائنا إلى واجهة برمجة تطبيقات يمكن لوكيل مضيف إرسال إشعارات إليها. بعد أن تمكّنت الوكلاء البعيدون من الوصول إلى خوادم A2A الخاصة بهم، علينا إنشاء وكيل مضيف يكتشفهم وينسّق عملهم.
هناك بعض الجوانب المختلفة للمضيف التي نحتاج إلى إنشائها لتحقيق ذلك. أولاً، علينا إنشاء طريقة لإنشاء برامج فردية لكل وكيل من وكلائنا البعيدين وخوادمهم. يتم ذلك من خلال إنشاء مصنع A2A Client الذي ينشئ اتصالات بكل نقاط نهاية الوكيل البعيد المستضافة على خوادمها، حيث يمكن للمضيف اكتشاف بطاقات الوكيل. يمكن تهيئة عمليات الربط بين المضيف وكل وكيل بعيد باستخدام عنصر RemoteAgentConnections.
touch $HOME/app/src/host/remote_agent_connection.py
touch $HOME/app/src/host/agent.py
touch $HOME/app/src/host/__init__.py
echo "from . import agent" >> $HOME/app/src/host/__init__.py
انسخ عملية التنفيذ التالية لعمليات ربط الوكيل البعيد في ملف src/host/remote_agent_connection.py:
# src/host/remote_agent_connection.py
import traceback
from a2a.client import (
Client,
ClientFactory,
)
from a2a.types import (
AgentCard,
Message,
SendMessageRequest,
Task,
TaskState,
)
class RemoteAgentConnection:
"""A class to hold the connections to the remote agents."""
def __init__(self, client_factory: ClientFactory, agent_card: AgentCard):
self.agent_client: Client = client_factory.create(agent_card)
self.card: AgentCard = agent_card
def get_agent(self) -> AgentCard:
return self.card
async def send_message(self, message: Message) -> Task | Message | None:
lastTask: Task | None = None
try:
async for event in self.agent_client.send_message(
SendMessageRequest(message=message)
):
if event.HasField('message'):
return event.message
# Build up the task from the streamed task and update events
if event.HasField('task'):
lastTask = event.task
elif lastTask and event.HasField('status_update'):
lastTask.status.CopyFrom(event.status_update.status)
elif lastTask and event.HasField('artifact_update'):
lastTask.artifacts.append(event.artifact_update.artifact)
if lastTask and self.is_terminal_or_interrupted(lastTask):
return lastTask
except Exception as e:
print('Exception found in send_message')
traceback.print_exc()
raise e
return lastTask
def is_terminal_or_interrupted(self, task: Task) -> bool:
return task.status.state in [
TaskState.TASK_STATE_COMPLETED,
TaskState.TASK_STATE_CANCELED,
TaskState.TASK_STATE_FAILED,
TaskState.TASK_STATE_REJECTED,
TaskState.TASK_STATE_INPUT_REQUIRED,
TaskState.TASK_STATE_AUTH_REQUIRED,
]
أصبح بإمكاننا الآن إنشاء روابط بين العميل والخادم من خلال بطاقة العميل وبطاقة الوكيل المحدّدة. ستكون هذه أداة يستخدمها وكيل المضيف لإنشاء اتصالات مع خوادم الوكيل البعيد.
لننتقل الآن إلى إنشاء الوكيل المضيف نفسه. ابدأ بنسخ مقطع الرمز التالي في الملف app/src/host/agent.py. هذه هي عملية تهيئة فئة Pythonic التي تتطلّب نقاط نهاية الوكيل البعيد وعميل httpx عادي. سيؤدي تهيئة فئة الوكيل هذه إلى إنشاء الاتصالات من خلال العناوين البعيدة المقدَّمة إليها.
import asyncio
import json
import os
import uuid
import httpx
from typing import Any
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import (
AgentCard,
Message,
Part,
Role,
Task,
TaskState,
)
from a2a.utils import TransportProtocol
from google.protobuf.json_format import MessageToDict
from google.adk import Agent
from google.adk.models import Gemini
from google.genai import types
from google.adk.agents.callback_context import CallbackContext
from google.adk.agents.readonly_context import ReadonlyContext
from google.adk.tools.tool_context import ToolContext
from dotenv import load_dotenv
from .remote_agent_connection import RemoteAgentConnection
load_dotenv()
######################################
# --- Coordinator Agent Definition ---
######################################
class CoordinatorAgent:
"""
The Coordinator agent.
This is the agent responsible for sending tasks to agents.
"""
def __init__(
self,
remote_agent_addresses: list[str],
http_client: httpx.AsyncClient,
):
self.http_client = http_client
self.remote_agent_addresses = remote_agent_addresses
self.client_factory = None
self.remote_agent_connections: dict[str, RemoteAgentConnection] = {}
self.cards: dict[str, AgentCard] = {}
self.agents: str = ''
الجانب التالي من وكيل المضيف هو إنشاء اتصالات بالوكلاء البعيدين. بدلاً من الاتصال في __init__ (الذي يتم تنفيذه عند استيراد الوحدة، قبل توفّر حلقة معالجة الأحداث)، يؤجّل الأسلوب ensure_initialized هذا العمل إلى حين استخدام الوكيل فعليًا. يتحقّق هذا الإجراء مما إذا تمّ إنشاء اتصالات من قبل، وإذا لم يكن الأمر كذلك، ينشئ عميل A2A ويتّصل بكلّ وكيل بعيد بالتوازي. انسخ هذا المقتطف مباشرةً أسفل المقتطف السابق في app/src/host/agent.py.
#####################################
# --- Remote Agent Initialization ---
#####################################
async def ensure_initialized(self):
if not self.remote_agent_connections:
config = ClientConfig(
httpx_client=self.http_client,
supported_protocol_bindings=[
TransportProtocol.JSONRPC,
TransportProtocol.HTTP_JSON,
],
)
self.client_factory = ClientFactory(config=config)
async with asyncio.TaskGroup() as task_group:
for address in self.remote_agent_addresses:
task_group.create_task(self.retrieve_card(address))
async def retrieve_card(self, address: str):
card_resolver = A2ACardResolver(self.http_client, address)
card = await card_resolver.get_agent_card()
# Use the actual address, not the hardcoded URL in the card JSON
for interface in card.supported_interfaces:
interface.url = address
remote_connection = RemoteAgentConnection(self.client_factory, card)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
agent_info = []
for card in self.cards.values():
agent_info.append(
json.dumps({"name": card.name, "description": card.description})
)
self.agents = "\n".join(agent_info)
بعد ذلك، لدينا تنفيذ LLM Agent. في هذه الحالة، نستخدم وكيل ADK لمنسّقنا، ما يتطلّب استخدام الإضافات النموذجية، مثل تعليمات الطلبات وعمليات معاودة الاتصال والأدوات. انسخ كل هذه المكوّنات داخل فئة وكيل المضيف في الملف app/src/host/agent.py. ضَع وظيفة معاودة الاتصال التالية في فئة وكيل المضيف. قبل أن يتم تشغيل الوكيل بشكل صحيح، تعمل عملية معاودة الاتصال هذه على تهيئة الوكيل إذا لم تتم تهيئته من قبل في حالة الوكيل.
############################
# -- Implement the ADK Agent
############################
# --- Before Model Callback ---
def before_model_callback(
self, callback_context: CallbackContext, llm_request
):
"""
A callback to set up the session state before the model processes the
request
"""
state = callback_context.state
if 'session_active' not in state or not state['session_active']:
if 'session_id' not in state:
state['session_id'] = str(uuid.uuid4())
state['session_active'] = True
المكوّن التالي لوكيل المضيف هو تعليمات الطلب. بما أنّ هذا هو وكيل التنسيق، يجب أن يعرف الوكلاء البعيدين الذين سيكونون تحت تصرّفه، بالإضافة إلى الوكيل النشط الحالي لمنح المنسّق فكرة عمّا يحدث حاليًا في الجلسة. أضِف الطلب ودالة المساعدة التالية أسفل دالة رد الاتصال.
# --- Prompt ---
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_state(context)
return f"""
You are an expert orchestrator that can delegate user requests to the
appropriate remote agents to generate a GitHub research report.
**Your Goal:** To fulfill user requests for GitHub repository data, evaluate it, and optionally email a report.
**Workflow Steps:**
1. **Understand the User Request**:
- Identify repository names (e.g., "google/adk-python").
- Determine if the user wants Pull Requests, Issues, or both.
- Extract any specified email address for sending the report (e.g., "user@example.com").
- Note the number of PRs/issues requested per repository. If not specified, the Retrieval Agent has defaults.
2. **Retrieve Data (GitHub_Retrieval_Agent)**:
- Use the `send_message` tool to send a message to the "GitHub_Retrieval_Agent".
- Your message to the Retrieval Agent should clearly state which repositories to fetch data for, and specify if you need issues, pull requests, and the respective limits.
- Example message to Retrieval Agent: "Fetch 5 issues and 3 pull requests for google/adk-python."
3. **Evaluate and Summarize Data (Content_Evaluation_Agent)**:
- Once you receive the raw JSON data from the "GitHub_Retrieval_Agent", use the `send_message` tool to send this data to the "Content_Evaluation_Agent".
- Your message to the Evaluation Agent should include the raw JSON data you received.
- The Evaluation Agent will return a Markdown-formatted report.
4. **Email Report (Email_Agent - if email provided)**:
- If the user's initial request included an email address, use the `send_message` tool to send the Markdown report from the "Content_Evaluation_Agent" to the "Email_Agent".
- Your message to the Email Agent should include the report and the recipient's email address.
- Example message to Email Agent: "Send this report to user@example.com: [Markdown Report Content]".
5. **Respond to User**:
- Based on the outcome of the steps above, formulate a concise and informative response to the user.
- If a report was generated and emailed, confirm that. If only a report was generated, provide it directly to the user.
- If an email address was requested but the email failed to send, inform the user.
- If any step failed, inform the user about the failure.
**Available Tools:**
- `list_remote_agents()`: Use this to see what agents are available (though you already know their names for this workflow).
- `send_message(agent_name: str, message: str)`: Use this to interact with remote agents.
**Crucial Guidelines:**
- **Rely on Tools**: ALWAYS use `send_message` to interact with the remote agents. Do NOT attempt to perform the tasks yourself.
- **No Conversational Filler**: Only communicate the essential information to the remote agents or back to the user.
- **Error Handling**: If a remote agent returns an error, acknowledge it and try to provide a helpful message to the user.
Agents:
{self.agents}
Current agent: {current_agent['active_agent']}
"""
def check_state(self, context: ReadonlyContext):
state = context.state
if (
'context_id' in state
and 'session_active' in state
and state['session_active']
and 'agent' in state
):
return {'active_agent': f'{state["agent"]}'}
return {'active_agent': 'None'}
الآن يأتي الجزء الأهم من الوكيل المضيف، وهو الأدوات. سيتمكّن المضيف من الوصول إلى أداة send_message() وأداة list_remote_agents(). أداة list_remote_agent() أبسط، فهي تقرأ بطاقات الوكيل الخاصة بالفئة وتعرض قائمة بالقواميس التي تحتوي على اسم كل وكيل ووصفه. الدالة send_message() أكثر تقدّمًا قليلاً. يمكنه إرسال رسالة أو مهمة أو بث أو بث غير مباشر إلى وكيل بعيد، وذلك باستخدام اسم الوكيل وسلسلة الرسالة. ضَع جزء الرمز هذا داخل فئة Python نفسها الخاصة بـ Host Agent أسفل قسم الطلب في app/src/host/agent.py.
# --- Agent Tools ---
def list_remote_agents(self):
"""List the available remote agents you can use to delegate the task."""
if not self.remote_agent_connections:
return []
remote_agent_info = []
for card in self.cards.values():
remote_agent_info.append(
{'name': card.name, 'description': card.description}
)
return remote_agent_info
async def send_message(
self, agent_name: str, message: str, tool_context: ToolContext
):
"""Sends a task either streaming (if supported) or non-streaming.
This will send a message to the remote agent named agent_name.
Args:
agent_name: The name of the agent to send the task to.
message: The message to send to the agent for the task.
tool_context: The tool context this method runs in.
Yields:
A dictionary of JSON data.
"""
await self.ensure_initialized()
if agent_name not in self.remote_agent_connections:
raise ValueError(f'Agent {agent_name} not found')
state = tool_context.state
state['agent'] = agent_name
client = self.remote_agent_connections[agent_name]
if not client:
raise ValueError(f'Client not available for {agent_name}')
task_id = state.get('task_id', None)
context_id = state.get('context_id', None)
message_id = state.get('message_id', None)
task: Task
if not message_id:
message_id = str(uuid.uuid4())
request_message = Message(
role=Role.ROLE_USER,
parts=[Part(text=message)],
message_id=message_id,
context_id=context_id,
task_id=task_id,
)
response = await client.send_message(request_message)
if isinstance(response, Message):
return await convert_parts(response.parts, tool_context)
task: Task = response # type: ignore
# Assume completion unless a state returns that isn't complete
state['session_active'] = not client.is_terminal_or_interrupted(task)
if task.context_id:
state['context_id'] = task.context_id
# Only reuse the task ID to continue a task that is waiting for input.
# Finished tasks can't take new messages, so the next call starts a new task.
if task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
state['task_id'] = task.id
else:
state['task_id'] = None
if task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
# Force user input back
tool_context.actions.skip_summarization = True
tool_context.actions.escalate = True
elif task.status.state == TaskState.TASK_STATE_CANCELED:
# Open question, should we return some info for cancellation instead
raise ValueError(f'Agent {agent_name} task {task.id} is cancelled')
elif task.status.state == TaskState.TASK_STATE_FAILED:
# Raise error for failure
raise ValueError(f'Agent {agent_name} task {task.id} failed')
response = []
if task.status.HasField('message'):
response.extend(
await convert_parts(task.status.message.parts, tool_context)
)
if task.artifacts:
for artifact in task.artifacts:
response.extend(
await convert_parts(artifact.parts, tool_context)
)
return response
أخيرًا، سننفّذ وكيل ADK الذي يتضمّن كل هذه المكوّنات. هذا هو الجزء الأساسي من وكيل فئة المنسّق الذي ينفّذ الطلب الذي تم تقديمه باستخدام الأدوات المتاحة له بعد تنفيذ BeforeModelCallback.
ضَع مقطع الرمز التالي أسفل قسم الأدوات مباشرةً في ملف app/src/host/agent.py أسفل قسم الأدوات:
def create_agent(self) -> Agent:
"""Create an instance of the CoordinatorAgent."""
return Agent(
model=Gemini(
model='gemini-3.8-flash',
retry_options=types.HttpRetryOptions(attempts=5),
),
name='host',
instruction=self.root_instruction,
before_model_callback=self.before_model_callback,
description=(
'This coordinator agent orchestrates the retriever, evaluator, and emailer agents'
),
tools=[
self.send_message,
self.list_remote_agents,
],
)
يحتاج الوكيل إلى بعض الدوال المساعدة لأداة send_message() من أجل تحويل أجزاء A2A إلى نص وبيانات أولية. انسخ هذا الجزء من الرمز البرمجي خارج فئة CoordinatorAgent:
##########################
# --- Helper Functions ---
##########################
async def convert_part(part: Part, tool_context: ToolContext):
"""Convert a part to text. Only text parts are supported."""
if part.HasField("text"):
return part.text
if part.HasField("data"):
return MessageToDict(part.data)
return f"Unknown type: {part}"
async def convert_parts(parts: list[Part], tool_context: ToolContext):
"""Convert parts to text."""
rval = []
for p in parts:
rval.append(await convert_part(p, tool_context))
return rval
لتسهيل الاختبار والتفاعل على الجهاز المحلي من خلال adk web أو adk run، أضِف عملية تهيئة root_agent في أسفل ملف app/src/host/agent.py:
root_agent = CoordinatorAgent(
remote_agent_addresses=[
os.getenv('RETRIEVAL_AGENT_URL', 'http://localhost:8001'),
os.getenv('EVALUATOR_AGENT_URL', 'http://localhost:8002'),
os.getenv('EMAILER_AGENT_URL', 'http://localhost:8003'),
],
http_client=httpx.AsyncClient(timeout=30),
).create_agent()
تهانينا! لقد أنشأت للتوّ وكيل المضيف الأول من تطبيق إلى تطبيق. بما أنّنا بدأنا تشغيله باستخدام المتغيّر root_agent، يمكننا التفاعل معه من خلال adk run أو عملية النشر المحلية adk web تمامًا مثل النماذج الوكيلة الأخرى عن بُعد. استخدِم الأوامر التالية للتفاعل مع المضيف:
cd $HOME/app/src/
uv run adk run host
لا يمكن للمضيف الردّ إلا بعد تشغيل خوادم الوكيل الثلاثة البعيدة، وهو ما ستفعله في الخطوة التالية. إلى حين ذلك، يمكن للمضيف الردّ على تحية، ولكن أي طلب يحتاج إلى وكيل عن بُعد سيتعذّر تنفيذه وسيظهر الخطأ AgentCardResolutionError. اكتب exit لإيقاف المضيف.
10. الاختبار محليًا
لاختبار نظام الوكلاء المتعدّدين الكامل، عليك بدء خوادم الوكلاء البعيدين التي تريد أن يتيحها الوكيل المضيف. سيبدأ هذا النص البرمجي جميع خوادم الوكلاء في الوقت نفسه. أنشئ ملف النص البرمجي:
touch $HOME/app/start_agents.sh
cloudshell edit $HOME/app/start_agents.sh
انسخ النص البرمجي التالي في $HOME/app/start_agents.sh:
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Function to stop all background processes
cleanup() {
echo "Stopping agent servers..."
kill $(jobs -p) 2>/dev/null
wait
echo "All processes terminated."
}
# Call the cleanup function when the script exits or is interrupted
trap cleanup EXIT
# Start the agent servers in the background from the app directory
cd $HOME/app
echo "Starting agent servers..."
uv run python3 -m src.agents.retriever.server &
uv run python3 -m src.agents.evaluator.server &
uv run python3 -m src.agents.emailer.server &
# Start the host agent (it reads the remote agent URLs from .env)
cd $HOME/app/src/
uv run adk run host
# Wait for all background processes to finish
wait
بعد ذلك، شغِّل النص البرمجي:
bash $HOME/app/start_agents.sh
سيبدأ هذا النص البرمجي جميع الوكلاء الأربعة (المسترجِع والمقيّم والمرسِل عبر البريد الإلكتروني والمضيف). ستظهر لك نتائج السجلّ من كل خادم في نافذة الوحدة الطرفية. بعد تشغيل الخوادم، يمكنك فتح نافذة طرفية جديدة وتشغيل واجهة adk web:
# In a new terminal
cd $HOME/app/src/
uv run adk web
افتح رابط localhost الذي يظهر في الوحدة الطرفية. يمكنك من هنا التفاعل مع وكيل المضيف مباشرةً من خلال اختياره من القائمة المنسدلة في أعلى يمين الصفحة.
11. تنظيف
لتجنُّب تحصيل رسوم مستمرة، احذف الموارد التي تم إنشاؤها أثناء هذا الدرس التطبيقي حول الترميز.
أوقِف جميع خوادم الوكيل النشطة، ثم احذف المشروع إذا أنشأت مشروعًا خصيصًا لهذا الدرس العملي. انتقِل إلى وحدات التحكّم التي شغّلت الوكلاء البعيدين ونفِّذ الأمر Ctrl+C، ثم أزِل مشروع Google Cloud هذا:
gcloud projects delete <YOUR_PROJECT_ID>
12. تهانينا
لقد أنشأت نظامًا متعدد الوكلاء بنجاح باستخدام Agent2Agent.
ما تعلّمته
- كيفية إنشاء وكلاء مستقلين في حزمة ADK باستخدام أدواتهم الخاصة
- كيفية منح الوكلاء هويات قابلة للاكتشاف باستخدام "بطاقات الوكلاء"
- كيفية إتاحة الوصول إلى الوكلاء من خلال خوادم التطبيقات إلى التطبيقات
- كيفية إنشاء وكيل مضيف ينسّق الوكلاء البعيدين
- كيف يتيح بروتوكول A2A التواصل بين النماذج الوكيلة التي تم تطويرها بشكل مستقل؟
الخطوات التالية
- نشر النظام على Cloud Run لاستخدامه في الإنتاج
- إضافة المزيد من الوكلاء عن بُعد لتوسيع نطاق إمكانات النظام
- استكشاف إمكانات بثّ المحتوى والإشعارات الفورية في A2A