1. Обзор
В этом практическом занятии вы создадите многоагентную систему, в которой несколько агентов ADK взаимодействуют и сотрудничают, используя протокол Agent2Agent (A2A).
Что вы узнаете
- Как создать несколько независимых агентов ADK
- Как выдать каждому агенту карточку агента и использовать их в качестве A2A-серверов
- Как создать хост-агент, который будет управлять вашими удалёнными агентами
- Как установить удаленное соединение с агентом
- Как протестировать многоагентную систему локально
Что вам понадобится
- Проект Google Cloud с включенной функцией выставления счетов.
- Веб-браузер, например Chrome.
- Python 3.12+
Этот практический урок предназначен для разработчиков среднего уровня, имеющих некоторое представление о Python и Google Cloud.
Выполнение этого практического задания займет приблизительно 15 минут.
Стоимость ресурсов, созданных в рамках этого практического занятия, должна составлять менее 5 долларов.
2. Настройте свою среду.
Создайте проект в Google Cloud.
- В консоли Google Cloud на странице выбора проекта выберите или создайте проект Google Cloud .
- Убедитесь, что для вашего облачного проекта включена функция выставления счетов. Узнайте, как проверить, включена ли функция выставления счетов для проекта .
Запустить редактор CloudShell
Чтобы запустить сеанс Cloud Shell из консоли Google Cloud, нажмите кнопку «Активировать Cloud Shell» в консоли Google Cloud.
Это запустит сессию в нижней панели вашей консоли Google Cloud.
Чтобы запустить редактор, нажмите кнопку «Открыть редактор» на панели инструментов окна Cloud Shell.
Настройте свою среду
Для начала выполните следующую команду в терминале, чтобы создать структуру папок проекта для вашей системы A2A. В целях этой демонстрации мы будем использовать абсолютные пути от вашего домашнего каталога ($HOME):
mkdir -p $HOME/app/src/agents $HOME/app/src/host
touch $HOME/app/.env $HOME/app/pyproject.toml
Теперь, когда у нас есть общая архитектура, давайте заполним конфигурации среды. Скопируйте следующий фрагмент кода в новый файл .env :
Заполните свой новый файл .env идентификатором проекта GCP и регионом GCP. Вы также можете указать свой адрес электронной почты в поле MAIL_TO , если хотите получать отчеты по электронной почте от агента, который вы собираетесь собрать. Кроме того, вы можете добавить персональный токен доступа GitHub GITHUB_TOKEN для упрощения работы агента получения данных из GitHub.
Откройте созданный файл .env с помощью следующей команды bash:
cloudshell edit .env
Затем скопируйте следующий файл в файл .env по адресу app.env . ВАЖНОЕ ПРИМЕЧАНИЕ: Убедитесь, что вы указали СВОИ собственные значения.
# --- Google Cloud ---
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT=<your-gcp-project-id>
GOOGLE_CLOUD_LOCATION=<your-gcp-project-region>
# --- 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>
MCP_SERVER_HOST=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 :
[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,<3.14"
dependencies = [
"resend>=2.21.0",
"uvicorn>=0.40.0",
"a2a-sdk>=0.3.26,<0.4.0",
"google-adk[a2a]>=1.27.0,<1.30.0",
"google-generativeai>=0.8.4",
"httpx>=0.28.1",
"pydantic>=2.12.5, <3.0.0",
"python-dotenv>=1.2.0",
"nest-asyncio>=1.6.0",
]
[project.optional-dependencies]
test = [
"pytest>=8.3.2",
"pytest-asyncio>=0.23.8",
"pytest-mock>=3.14.0",
"respx>=0.21.0",
]
[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 , запустите в терминале следующий bash-скрипт. Это настроит вашу виртуальную среду Python и установит все необходимые зависимости из файла pyproject.toml .
# Ensure you are in the 'app' directory before running this
# Exit immediately if a command exits with a non-zero status.
set -e
# 1. Install uv, the Python package manager used for this project
echo "Installing uv..."
if ! command -v uv &> /dev/null
then
pip install uv
else
echo "uv is already installed."
fi
# 2. Create and activate virtual environment
echo "Setting up virtual environment..."
# --clear ensures you start with a fresh environment
uv venv --clear
# 3. Install dependencies
echo "Installing Python dependencies..."
uv sync
echo "Installation and setup complete."
Теперь у нас есть все необходимое для создания нашей многоагентной системы!
3. Создайте агента-извлекателя.
Ева — разработчик программного обеспечения, которая хочет быть в курсе обновлений репозиториев 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.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-2.5-flash",
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",
},
)
)
],
)
Этот агент работает как автономный инструмент. Для его независимого тестирования выполните следующие команды: `uv run adk web` в каталоге `**/agents/`:
cd $HOME/app/src/agents
uv run adk run retriever
Откройте ссылку localhost в терминале 'uv run adk web' и выберите агента, чтобы попробовать его в действии.
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 google.adk.agents.llm_agent import Agent
# --- 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-2.5-flash",
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.
""",
)
Этот агент работает как автономный инструмент. Для его независимого тестирования выполните следующие команды в НОВОМ терминале: `uv run adk web` В КАТАЛОГЕ **/agents/:
cd $HOME/app/src/agents
uv run adk run evaluator
Откройте ссылку localhost в терминале и выберите агента-оценщика, чтобы попробовать его в действии.
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 resend
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
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 = f"<div style='font-family: sans-serif; white-space: pre-wrap;'>{body}</div>"
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-2.5-flash",
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
],
)
Этот агент работает как автономный инструмент. Для его независимого тестирования выполните следующие команды в НОВОМ терминале: `uv run adk web` В КАТАЛОГЕ **/agents/:
cd $HOME/app/src/agents
uv run adk run emailer
Откройте ссылку localhost в терминале и выберите агента, чтобы попробовать его в работе.
6. Создайте карточки агентов.
Мы только что рассмотрели три истории независимых разработчиков, у каждого из которых есть свои уникальные агенты. Все эти агенты опубликованы и зарегистрированы в библиотеке агентов их компаний. Другой разработчик, Диана, хочет объединить эти отдельные идеи в единую многоагентную систему.
Ее цель — получить доступ к поисковому агенту, который будет оперативно информировать ее о проблемах и запросах на слияние (PR) в различных средах разработки агентов. Она также хочет, чтобы агент отправлял ей по электронной почте краткое резюме всех новых разработок в экосистеме. Поскольку все агенты разрабатываются индивидуально в разных средах, A2A является идеальным решением для объединения этих существующих агентов.
Первым шагом в создании группы удалённых агентов является присвоение каждому из необходимых удалённых агентов карточки агента. Представьте их как визитные карточки для вашего агента. Это JSON-файлы, используемые для того, чтобы ваш хост-агент мог идентифицировать агентов по их имени, описанию, возможностям и схемам ввода/вывода.
Сначала выполните следующую команду из корневой директории вашего проекта, чтобы создать каталог 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.",
"url": "http://localhost:8001",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"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.",
"url": "http://localhost:8002",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"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.",
"url": "http://localhost:8003",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"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, который упрощает отправку сообщений и упорядочивание задач удаленным агентам, обменивающимся артефактами TextPart. Благодаря этому мы можем использовать общий исполнитель агентов, используемый несколькими агентами. Создайте новый файл исполнителя:
touch $HOME/app/src/agents/executor.py
Скопируйте следующий фрагмент кода в новый файл $HOME/app/src/agents/executor.py :
# $HOME/app/src/agents/executor.py
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, Part
from a2a.utils import new_agent_text_message, new_task
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(context.message) # type: ignore
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, task.context_id)
if context.call_context:
user_id = context.call_context.user.user_name
else:
user_id = "a2a_user"
try:
# Update status with custom message
await updater.update_status(
TaskState.working,
new_agent_text_message(
self.status_message,
task.context_id,
task.id
),
)
# Process with ADK agent
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(root=TextPart(text=response_text))],
name=self.artifact_name,
)
await updater.complete()
except Exception as e:
await updater.update_status(
TaskState.failed,
new_agent_text_message(f"Error: {e!s}", task.context_id, task.id),
final=True,
)
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 на карточках агентов.
Следующий шаг для каждого из удалённых агентов — предоставление им доступа к собственному 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 a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
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 = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=RetrieverAgentExecutor(
agent=retriever_agent
),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=github_retrieval_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8001))
print(f"Starting Retriever Agent on Port {port}...")
uvicorn.run(server.build(), host="0.0.0.0", port=port)
Затем создайте сервер для ознакомительного агента:
# $HOME/app/src/agents/evaluator/server.py
import logging
import os
import json
import uvicorn
from a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
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 = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=EvaluatorAgentExecutor(agent=evaluator_agent),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=content_evaluator_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8002))
print(f"Starting Evaluator Agent on Port {port}...")
uvicorn.run(server.build(), host="0.0.0.0", port=port)
И наконец, для агента по рассылке электронных писем:
# $HOME/app/src/agents/emailer/server.py
import logging
import os
import json
import uvicorn
from a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
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 = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=EmailerAgentExecutor(
agent=emailer_agent
),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=emailer_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8003))
print(f"Starting Emailer Agent on Port {port}...")
uvicorn.run(server.build(), host="0.0.0.0", port=port)
9. Создайте хост-агент
До настоящего момента мы, по сути, превратили каждого из наших агентов в API для отправки пингов хост-агенту. Теперь, когда наши удалённые агенты получили доступ к своим собственным A2A-серверам, нам необходимо создать хост-агент, который будет обнаруживать и координировать их работу.
Для этого нам необходимо разработать несколько различных аспектов хоста. Во-первых, нам нужно создать способ создания отдельных клиентов для каждого из наших удаленных агентов и их серверов. Это делается путем создания фабрики клиентов A2A, которая устанавливает соединения с каждой из точек доступа удаленных агентов, размещенных на их серверах, где хост может обнаружить карточки агентов. Соединения между хостом и каждым удаленным агентом можно инициализировать с помощью объекта 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,
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(message):
if isinstance(event, Message):
return event
if self.is_terminal_or_interrupted(event[0]):
return event[0]
lastTask = event[0]
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.completed,
TaskState.canceled,
TaskState.failed,
TaskState.input_required,
TaskState.unknown,
]
Теперь у нас есть возможность устанавливать соединения между клиентом и сервером с помощью заданного клиентского приложения и агентской карты. Это будет инструмент, который хост-агент будет использовать для установления соединений с удаленными агентскими серверами.
Перейдём к созданию самого хост-агента. Для начала скопируйте следующий фрагмент кода в файл app/src/host/agent.py . Это инициализация класса Python, которому требуются удалённые точки доступа агента и стандартный 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,
TextPart,
TransportProtocol,
)
from google.adk import Agent
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_transports=[
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()
card.url = address # Use the actual address, not the hardcoded URL in the card JSON
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. В данном случае мы используем агент 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() немного сложнее. Она позволяет отправлять сообщение или задачу, потоковым или непотоковым способом, удалённому агенту, зная его имя и строку сообщения. Поместите этот фрагмент кода внутрь того же класса Host Agent на Python, в раздел prompt в файле 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.user,
parts=[Part(root=TextPart(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
state['task_id'] = task.id
if task.status.state == TaskState.input_required:
# Force user input back
tool_context.actions.skip_summarization = True
tool_context.actions.escalate = True
elif task.status.state == TaskState.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.failed:
# Raise error for failure
raise ValueError(f'Agent {agent_name} task {task.id} failed')
response = []
if task.status.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.
Вставьте следующий фрагмент кода сразу после раздела tools в файле app/src/host/agent.py в подраздел tools:
def create_agent(self) -> Agent:
"""Create an instance of the CoordinatorAgent."""
return Agent(
model='gemini-2.5-flash',
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 isinstance(part.root, TextPart):
return part.root.text
if part.root.kind == "data":
return part.root.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
Для упрощения локального тестирования и взаимодействия через uv run 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()
ПОЗДРАВЛЯЕМ! Вы только что создали свой первый агент A2A на хосте. Поскольку мы инициализировали его с помощью переменной root_agent, мы можем взаимодействовать с ним через локальное развертывание ADK ADK, как и с другими удаленными агентами. Используйте следующие команды для взаимодействия с вашим хостом:
cd $HOME/app/src/
uv run adk run host
10. Протестируйте локально.
Для тестирования всей многоагентной системы необходимо запустить удаленные агентские серверы, которые вы хотите предоставить в распоряжение основного агента. Это запустит все агентские серверы одновременно.
# Make sure you have activated your virtual environment first!
# source .venv/bin/activate
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Function to kill all background processes
cleanup() {
echo "Caught signal, terminating background processes..."
# The negative PID kills the entire process group
kill -TERM -$$
wait
echo "All processes terminated."
exit
}
# Trap TERM, INT, and EXIT signals and call the cleanup function
trap cleanup TERM INT EXIT
# Activate virtual environment
uv sync
# Start the agent servers in the background
echo "Starting agent servers..."
uv run python3 -m src.agents.retriever.server --port 8001 &
uv run python3 -m src.agents.evaluator.server --port 8002 &
uv run python3 -m src.agents.emailer.server --port 8003 &
cd $HOME/app/src/
RETRIEVER_AGENT_URL=http://localhost:8001
EVALUATOR_AGENT_URL=http://localhost:8002
EMAILER_AGENT_URL=http://localhost:8003
uv run adk run host
# Wait for all background processes to finish
wait
Эта команда bash запустит все четыре сервера-агента (retriever, evaluator, emailer и host). Вы увидите вывод логов с каждого сервера в терминале. После запуска серверов вы можете открыть новое окно терминала (убедившись, что вы находитесь в том же каталоге app и виртуальная среда активирована) и запустить веб-интерфейс uv run adk:
# In a new terminal, from the 'app' directory
source .venv/bin/activate
cd $HOME/app/src/
uv run adk web
Откройте ссылку localhost, которая появится в вашем терминале. Отсюда вы можете напрямую взаимодействовать с агентом хоста, выбрав его из выпадающего меню в левом верхнем углу.
11. Уборка
Во избежание дальнейших списаний средств удалите ресурсы, созданные в ходе этого практического занятия.
Остановите все запущенные серверы агентов, затем удалите проект, если вы создали его специально для этого практического занятия. Перейдите в терминал, на котором были запущены удаленные агенты, нажмите Ctrl+C и удалите этот проект Google Cloud:
gcloud projects delete ${GOOGLE_CLOUD_PROJECT}
12. Поздравляем!
Вы успешно создали многоагентную систему с помощью Agent2Agent!
Что вы узнали
- Как создать независимых агентов ADK со своими собственными инструментами
- Как наделить агентов доступными для поиска идентификаторами с помощью карточек агентов.
- Как обеспечить доступ к агентам через A2A-серверы
- Как создать хост-агент, который управляет удаленными агентами
- Как протокол A2A обеспечивает связь между независимо разработанными агентами
Следующие шаги
- Разверните систему в Cloud Run для использования в производственной среде.
- Добавьте больше удаленных агентов, чтобы расширить возможности системы.
- Изучите возможности потоковой передачи и push-уведомлений в A2A.