1. Übersicht
In diesem Codelab erstellen Sie ein Multi-Agenten-System, in dem mehrere ADK-Agenten über das Agent2Agent-Protokoll (A2A) kommunizieren und zusammenarbeiten.
Lerninhalte
- Mehrere unabhängige ADK-Agents erstellen
- So geben Sie jedem Agenten eine Agentenkarte und verpacken ihn als A2A-Server
- Host-Agenten erstellen, die Ihre Remote-Agenten orchestrieren
- Remote-Agent-Verbindungen herstellen
- Multi-Agenten-System lokal testen
Voraussetzungen
- Ein Google Cloud-Projekt mit aktivierter Abrechnungsfunktion
- Ein Webbrowser wie Chrome
- Python 3.12+ und uv, wenn Sie den Code auf Ihrem eigenen Computer anstelle von Cloud Shell ausführen
Dieses Codelab richtet sich an Entwickler mit mittleren Kenntnissen, die mit Python und Google Cloud vertraut sind.
Dieses Codelab dauert etwa 15 Minuten.
Die in diesem Codelab erstellten Ressourcen sollten weniger als 5 $ kosten.
2. Umgebung einrichten
Google Cloud-Projekt erstellen
- Wählen Sie in der Google Cloud Console auf der Seite zur Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
- Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.
Cloud Shell-Editor starten
Klicken Sie zum Starten einer Cloud Shell-Sitzung in der Google Cloud Console auf Cloud Shell aktivieren.
Dadurch wird im unteren Bereich der Google Cloud Console eine Sitzung gestartet.
Klicken Sie zum Starten des Editors in der Symbolleiste des Cloud Shell-Fensters auf Editor öffnen.
APIs aktivieren
Führen Sie im Terminal den folgenden Befehl aus, um die API zu aktivieren, die vom ADK zum Aufrufen von Gemini verwendet wird:
gcloud services enable aiplatform.googleapis.com
Umgebung konfigurieren
Führen Sie zuerst den folgenden Befehl in Ihrem Terminal aus, um die Projektordnerstruktur für Ihr A2A-System zu erstellen. Für diese Demo verwenden wir absolute Pfade aus Ihrem $HOME-Verzeichnis:
mkdir -p $HOME/app/src/agents $HOME/app/src/host
touch $HOME/app/.env $HOME/app/pyproject.toml
Nachdem wir die allgemeine Architektur haben, können wir die Umgebungskonfigurationen einfügen. Kopieren Sie das folgende Code-Segment in die neue Datei .env:
Fügen Sie in die neue Datei .env Ihre Google Cloud-Projekt-ID ein. Lassen Sie GOOGLE_CLOUD_LOCATION auf global eingestellt, wo das Gemini-Modell in diesem Codelab verfügbar ist. Sie können auch eine beliebige E‑Mail-Adresse in das MAIL_TO eingeben, wenn Sie Berichts-E‑Mails von dem Agenten erhalten möchten, den Sie erstellen. Sie können kontinuierlich ein persönliches GitHub-Zugriffstoken GITHUB_TOKEN hinzufügen, um den GitHub-Abruf-Agenten zu unterstützen.
Öffnen Sie die neue .env-Datei mit dem folgenden Bash-Befehl:
cloudshell edit $HOME/app/.env
Kopieren Sie dann die folgende Datei in die .env-Datei unter app/.env. WICHTIGER HINWEIS: Achten Sie darauf, dass Sie IHRE eigenen Werte eingeben.
# --- 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
Nachdem Sie die Umgebungsvariablen festgelegt haben, müssen wir die uv-Umgebung konfigurieren. Öffnen Sie die Datei pyproject.toml.
cloudshell edit $HOME/app/pyproject.toml
Kopieren Sie das folgende Code-Segment in die Datei 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/"]
Abhängigkeiten installieren
Führen Sie nun die folgenden Befehle in Ihrem Terminal aus, um in das gerade erstellte Verzeichnis app zu wechseln und alle erforderlichen Abhängigkeiten aus der Datei pyproject.toml zu installieren:
cd $HOME/app
uv sync
uv erstellt eine isolierte Python-Umgebung für dieses Codelab, sodass Sie nichts aktivieren müssen. Stellen Sie Python-Befehlen das Präfix uv run voran.
Jetzt haben wir alles, was wir zum Erstellen unseres Multi-Agenten-Systems benötigen.
3. Retriever-Agent erstellen
Eva ist Softwareentwicklerin und möchte über GitHub-Repositories auf dem Laufenden bleiben, wenn sie mit neuen Problemen und Pull-Anfragen aktualisiert werden. Sie erstellt also einen ADK-Agenten, um die angeforderten GitHub-Daten abzurufen.
Führen Sie für diese Demo den folgenden Befehl im Stammverzeichnis Ihres Projekts aus, um das erforderliche Verzeichnis und die erforderliche Datei für den Retriever-Agenten zu erstellen:
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
Kopieren Sie das folgende ADK-Code-Segment in $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",
},
)
)
],
)
Dieser KI-Agent funktioniert als eigenständiges Tool. Wenn Sie den Agenten unabhängig testen möchten, führen Sie die folgenden Befehle aus, um mit ihm über adk run aus dem Verzeichnis agents zu chatten:
cd $HOME/app/src/agents
uv run adk run retriever
Geben Sie an der Eingabeaufforderung [user]: eine Nachricht ein, um den Agent auszuprobieren, und geben Sie dann exit ein, um den Agent zu beenden.
4. Evaluator-Agent erstellen
Richard muss oft viel Fachjargon für seine nicht technischen Kunden und Kollegen vereinfachen. Richard hat es satt, immer wieder dieselben Begriffe zu definieren und dasselbe Projekt in einer Kurzfassung zu erklären. Deshalb entwickelt er einen Agenten, der technische Nomenklatur in leicht verständlichen Text zusammenfasst.
Führen Sie für diese Demo den folgenden Befehl aus, um die Datei für den Evaluator-Agenten zu erstellen:
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
Kopieren Sie das folgende Agent-Code-Segment in $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.
""",
)
Dieser KI-Agent funktioniert als eigenständiges Tool. Wenn Sie es unabhängig testen möchten, führen Sie die folgenden Befehle in einem NEUEN Terminal aus, um über das Verzeichnis agents mit adk run zu chatten:
cd $HOME/app/src/agents
uv run adk run evaluator
Geben Sie an der Eingabeaufforderung [user]: eine Nachricht ein, um den Agent auszuprobieren, und geben Sie dann exit ein, um den Agent zu beenden.
5. Emailer-Agent erstellen
Ivan hat es satt, so viele E‑Mails schreiben zu müssen, in denen er nur einen leicht verfügbaren Text zusammenfasst und neu formatiert. Er hat also einen E‑Mail-Agenten erstellt, der einen bestimmten Text neu formatiert und an ein angegebenes E‑Mail-Konto sendet.
Führen Sie für diese Demo den folgenden Befehl in Ihrem Terminal aus, um die Datei für den E-Mail-Agenten zu erstellen:
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
Kopieren Sie das folgende Agent-Code-Segment in 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
],
)
Dieser KI-Agent funktioniert als eigenständiges Tool. Wenn Sie es unabhängig testen möchten, führen Sie die folgenden Befehle in einem NEUEN Terminal aus, um über das Verzeichnis agents mit adk run zu chatten:
cd $HOME/app/src/agents
uv run adk run emailer
Geben Sie an der Eingabeaufforderung [user]: eine Nachricht ein, um den Agent auszuprobieren, und geben Sie dann exit ein, um den Agent zu beenden.
6. Agentenkarten erstellen
Wir haben uns gerade drei unabhängige Entwicklergeschichten angesehen, die alle ihre eigenen einzigartigen Agents haben. Alle diese Agents sind veröffentlicht und in der Agent-Bibliothek des jeweiligen Unternehmens angemeldet. Eine andere Entwicklerin, Diane, möchte diese einzelnen Ideen in einem einzigen Multi-Agenten-System zusammenführen.
Ihr Ziel ist es, einen On-Demand-Forschungs-KI-Agenten zu haben, der sie über die Probleme und PRs aus verschiedenen Frameworks für die Entwicklung von KI-Agenten auf dem Laufenden hält. Außerdem soll sie eine E‑Mail mit einer Zusammenfassung aller neuen Entwicklungen im Ökosystem senden. Da alle Agents einzeln in verschiedenen Umgebungen entwickelt werden, ist A2A eine ideale Lösung, um diese vorhandenen Agents zusammenzuführen.
Der erste Schritt beim Zusammenstellen einer Reihe von Remote-Agents besteht darin, jedem der erforderlichen Remote-Agents eine Agentenkarte zuzuweisen. Sie können sich diese als Visitenkarten für Ihren KI-Agenten vorstellen. Es handelt sich um JSON-Dateien, mit denen Ihr Host-Agent Agenten anhand ihres Namens, ihrer Beschreibungen, ihrer Funktionen und ihrer Ein-/Ausgabe-Schemas identifizieren kann. Die Karten in diesem Codelab verwenden das A2A v1.0-Agentenkartenformat, in dem jeder Endpunkt aufgeführt ist, über den der Agent in supportedInterfaces erreicht werden kann.
Führen Sie zuerst den folgenden Befehl im Stammverzeichnis Ihres Projekts aus, um das Verzeichnis cards und alle erforderlichen JSON-Dateien für die Agentenkarten zu erstellen:
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
Auf diese Dateien kann nur über den folgenden Bash-Befehl zugegriffen werden:
cloudshell edit $HOME/app/cards/github_retrieval_agent_card.json
Kopieren Sie den folgenden JSON-Inhalt in die Datei 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."
]
}
]
}
Kopieren Sie den folgenden Inhalt in die Datei app/cards/content_evaluator_agent_card.json. So erhält der Host-Agent eine Beschreibung der Funktionen dieses Agents, der Art der Daten, die Sie ihm zur Verfügung stellen können, und der Art der Daten, die er zurückgibt.
Verwenden Sie diesen Befehl wie zuvor, um die Evaluator-Agent-Karte zu bearbeiten:
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."
]
}
]
}
Hier ist die Agentenkarte für den E-Mail-Agenten. Kopieren Sie es mit demselben Cloud Shell-Befehl in die Datei app/cards/emailer_agent_card.json:
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. Remote-Agent-Executors erstellen
Damit ein Host-Agent die Remote-Agents „aufrufen“ oder mit ihnen „kommunizieren“ kann, benötigt jeder einen Agent-Executor, der für den A2A-Server sichtbar ist und durch die Agentenkarte identifiziert wird. Der Executor ist eine abstrakte Klasse mit den Methoden „execute()“ und „cancel()“, die den Remote-Agent aufrufen und ihm eine Aufgabe oder Nachricht anbieten oder die Aufgabe vollständig abbrechen.
Hier ist eine Implementierung eines abstrakten ADK-Agent-Executors, der das Senden von Nachrichten und das Anordnen von Aufgaben an Remote-Agents erleichtert, wobei alle Part-Artefakte ausgetauscht werden. Daher können wir einen gemeinsamen Agent Executor verwenden, der von mehreren Agents genutzt wird. Erstellen Sie eine neue Executor-Datei:
touch $HOME/app/src/agents/executor.py
Kopieren Sie das folgende Code-Segment in die neue Datei $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")
Nachdem wir diesen abstrakten Agent Executor haben, können wir die Python-Vererbung verwenden, um den Executor an die spezifischen Funktionen und Anforderungen jedes Agenten anzupassen. In unserem Fall ist die Ausführung des Aufrufs des Agents mit einer Nutzlast, das Warten auf eine Antwort und das Abrufen der Antwortnutzlast für unseren Workflow universell. Deshalb sind keine besonderen Änderungen erforderlich. Für jeden der A2A-Server des Agenten ist jedoch ein Agent Executor erforderlich. Führen Sie den folgenden Befehl aus, um die Executor-Dateien für alle drei Agents zu erstellen:
touch $HOME/app/src/agents/retriever/executor.py
touch $HOME/app/src/agents/evaluator/executor.py
touch $HOME/app/src/agents/emailer/executor.py
Fügen Sie nun jeder Datei die entsprechenden Inhalte hinzu.
# $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. Remote-Agents für A2A-Server verfügbar machen
Da jeder Agent jetzt eine eigene Visitenkarte hat, können sie gefunden werden, sobald sie über einen A2A-Server an einen Endpunkt weitergeleitet werden. In diesem Codelab wird der gesamte Prozess lokal ausgeführt. Dazu werden Localhosts für jeden der Remote-Agent-Endpunkte verwendet. In der Praxis können sie jedoch an jedem beliebigen Endpunkt verfügbar gemacht werden und sind auffindbar, solange sie im url des Felds supportedInterfaces der Agentenkarten referenziert werden.
Jeder Server stellt seine Agentenkarte unter /.well-known/agent-card.json bereit, wo A2A-Clients danach suchen, und verarbeitet A2A JSON-RPC-Anfragen unter /.
Im nächsten Schritt werden die einzelnen Remote-Agents ihrem eigenen A2A-Server zur Verfügung gestellt. Führen Sie diesen Befehl aus, um die Datei server.py für alle drei Remote-Agents zu erstellen:
touch $HOME/app/src/agents/retriever/server.py
touch $HOME/app/src/agents/evaluator/server.py
touch $HOME/app/src/agents/emailer/server.py
Jetzt fügen Sie den Servercode für jeden Agent hinzu. Beginnen wir mit dem Retriever-Agent:
# $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)
Erstellen Sie dann den Server für den Evaluierungs-Agent:
# $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)
Und schließlich für den E-Mail-Agenten:
# $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. Host-Agent erstellen
Wir haben im Grunde jeden unserer Agents in eine API umgewandelt, die von einem Host-Agent angepingt werden kann. Nachdem unsere Remote-Agents ihre eigenen A2A-Server kennengelernt haben, müssen wir einen Host-Agenten erstellen, der sie erkennt und koordiniert.
Dafür müssen wir einige Aspekte des Hosts entwickeln. Zuerst müssen wir eine Möglichkeit schaffen, individuelle Clients für jeden unserer Remote-Agents und deren Server zu erstellen. Dazu wird eine A2A-Client-Factory erstellt, die Verbindungen zu den einzelnen Remote-Agent-Endpunkten herstellt, die auf den Servern des Hosts gehostet werden. Dort kann der Host die Agent-Karten finden. Die Verbindungen zwischen dem Host und den einzelnen Remote-Agents können mit einem RemoteAgentConnections-Objekt initialisiert werden.
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
Kopieren Sie die folgende Implementierung von Remote-Agent-Verbindungen in die Datei 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,
]
Wir können jetzt Verbindungen zwischen einem Client und einem Server über eine bestimmte Client- und eine bestimmte Agent-Karte herstellen. Dies ist ein Tool, mit dem der Host-Agent Verbindungen zu den Remote-Agent-Servern herstellt.
Als Nächstes erstellen wir den Host-Agenten. Kopieren Sie zuerst das folgende Code-Segment in die Datei app/src/host/agent.py. Dies ist die Initialisierung der Python-Klasse, für die die Endpunkte des Remote-Agents und ein Standard-httpx-Client erforderlich sind. Durch die Initialisierung dieser Agent-Klasse werden die Verbindungen über die bereitgestellten Remote-Adressen hergestellt.
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 = ''
Der nächste Aspekt des Host-Agents ist das Herstellen von Verbindungen zu den Remote-Agents. Anstatt eine Verbindung in __init__ herzustellen (was beim Importieren des Moduls geschieht, bevor eine Ereignisschleife vorhanden ist), verschiebt die Methode ensure_initialized diese Aufgabe auf den Zeitpunkt, zu dem der Agent tatsächlich verwendet wird. Es wird geprüft, ob bereits Verbindungen hergestellt wurden. Wenn nicht, wird der A2A-Client erstellt und es wird parallel eine Verbindung zu jedem Remote-Agent hergestellt. Kopieren Sie dieses Segment direkt unter das vorherige Segment in 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)
Als Nächstes folgt die Implementierung des LLM-Agents. In diesem Fall verwenden wir einen ADK-Agenten für unseren Orchestrator, der die typischen Add-ons wie Prompts, Callbacks und Tools erfordert. Kopieren Sie alle diese Komponenten in die Host-Agent-Klasse in der Datei app/src/host/agent.py. Fügen Sie den folgenden Callback in die Host-Agent-Klasse ein. Bevor der Agent richtig ausgelöst wird, initialisiert dieser Callback den Agenten, falls er noch nicht im Status des Agenten initialisiert wurde.
############################
# -- 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
Die nächste Komponente unseres Host-Agents ist die Prompts-Anweisung. Da es sich um unseren Orchestrator-Agenten handelt, muss er die Remote-Agenten kennen, die ihm zur Verfügung stehen, sowie den aktuellen aktiven Agenten, damit der Koordinator eine Vorstellung davon bekommt, was gerade in der Sitzung passiert. Fügen Sie den folgenden Prompt und die folgende Hilfsfunktion unter dem Callback ein.
# --- 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'}
Nun kommt der wichtigste Teil des Host-Agents: die Tools. Der Host hat Zugriff auf ein send_message()-Tool und ein list_remote_agents()-Tool. Das Tool „list_remote_agent()“ ist einfacher. Es liest nur die Agentenkarten der Klasse ein und gibt eine Liste von Wörterbüchern mit dem Namen und der Beschreibung jedes Agenten zurück. Die Funktion send_message() ist etwas komplexer. Es kann eine Nachricht oder Aufgabe (Streaming oder nicht Streaming) an einen Remote-Agenten senden, wenn der Agentenname und der Nachrichtenstring angegeben sind. Fügen Sie dieses Code-Segment in derselben Host-Agent-Python-Klasse unter dem Prompt-Abschnitt in app/src/host/agent.py ein.
# --- 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
Schließlich implementieren wir den ADK-Agenten, der alle diese Komponenten enthält. Dies ist das Gehirn des Orchestrator-Klassen-Agents, der den erhaltenen Prompt mit den ihm zur Verfügung stehenden Tools ausführt, nachdem der BeforeModelCallback ausgeführt wurde.
Fügen Sie das folgende Code-Segment direkt unter dem Abschnitt „tools“ in der Datei app/src/host/agent.py ein:
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,
],
)
Der Agent benötigt einige Hilfsfunktionen für das Tool „send_message()“, um A2A-Teile in Rohdaten und ‑text zu konvertieren. Kopieren Sie dieses Segment AUSSERHALB der CoordinatorAgent-Klasse:
##########################
# --- 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
Um lokale Tests und die Interaktion über adk web oder adk run zu ermöglichen, fügen Sie unten in der Datei app/src/host/agent.py eine root_agent-Initialisierung hinzu:
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()
Das wars auch schon! Sie haben gerade Ihren ersten A2A-Host-Agenten erstellt. Da wir ihn mit der Variable „root_agent“ initialisiert haben, können wir über adk run oder die lokale adk web-Bereitstellung mit ihm interagieren, genau wie mit den anderen Remote-Agents. Verwenden Sie die folgenden Befehle, um mit Ihrem Host zu interagieren:
cd $HOME/app/src/
uv run adk run host
Der Host kann erst antworten, wenn die drei Remote-Agent-Server ausgeführt werden. Das ist der nächste Schritt. Bis dahin kann der Host auf eine Begrüßung antworten, aber jede Anfrage, für die ein Remote-Agent erforderlich ist, schlägt mit einem AgentCardResolutionError fehl. Geben Sie exit ein, um den Host zu beenden.
10. Lokal testen
Wenn Sie das gesamte Multi-Agenten-System testen möchten, müssen Sie die Remote-Agent-Server starten, die Ihr Host-Agent zur Verfügung haben soll. Mit diesem Script werden alle Agent-Server gleichzeitig gestartet. Erstellen Sie die Skriptdatei:
touch $HOME/app/start_agents.sh
cloudshell edit $HOME/app/start_agents.sh
Kopieren Sie das folgende Script in $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
Führen Sie dann das Script aus:
bash $HOME/app/start_agents.sh
Mit diesem Script werden alle vier KI-Agenten (Retriever, Evaluator, E-Mailer und Host) gestartet. Im Terminal wird die Logausgabe der einzelnen Server angezeigt. Sobald die Server ausgeführt werden, können Sie ein neues Terminalfenster öffnen und die adk web-Benutzeroberfläche starten:
# In a new terminal
cd $HOME/app/src/
uv run adk web
Öffnen Sie den Localhost-Link, der im Terminal angezeigt wird. Hier können Sie direkt mit dem Host-Agenten interagieren, indem Sie ihn oben links im Drop-down-Menü auswählen.
11. Bereinigen
Löschen Sie die in diesem Codelab erstellten Ressourcen, um laufende Gebühren zu vermeiden.
Beenden Sie alle laufenden Agent-Server und löschen Sie dann das Projekt, falls Sie eines speziell für dieses Codelab erstellt haben. Wechseln Sie zu den Terminals, in denen die Remote-Agents gestartet wurden, führen Sie Ctrl+C aus und entfernen Sie dann dieses Google Cloud-Projekt:
gcloud projects delete <YOUR_PROJECT_ID>
12. Glückwunsch
Sie haben erfolgreich ein Multi-Agenten-System mit Agent2Agent erstellt.
Das haben Sie gelernt
- Unabhängige ADK-Agenten mit eigenen Tools erstellen
- Agenten mit Agentenkarten eine auffindbare Identität geben
- Agenten über A2A-Server bereitstellen
- Host-Agent erstellen, der Remote-Agents orchestriert
- So ermöglicht das A2A Protocol die Kommunikation zwischen unabhängig entwickelten Agenten
Nächste Schritte
- System für die Produktion in Cloud Run bereitstellen
- Fügen Sie weitere Remote-Agents hinzu, um die Funktionen des Systems zu erweitern.
- Streaming- und Push-Benachrichtigungsfunktionen in A2A