Agent-to-Agent Protocol (A2A) Setup for Inter-Agent Communication

Agent-to-Agent Protocol (A2A) Setup

Agent-to-Agent Protocol (A2A) — открытый стандарт Google для взаимодействия между AI-агентами разных систем и провайдеров. A2A определяет, как агент-клиент (the caller) может обнаружить возможности агента-сервера (the remote agent), делегировать ему задачи и получать результаты — независимо от того, на каком фреймворке реализован каждый агент. Дополняет MCP (доступ к инструментам) протоколом агент-агент коммуникации.

Ключевые концепции A2A

Agent Card: JSON-документ, описывающий возможности агента (аналог OpenAPI для агентов). Публикуется на /.well-known/agent.json.

Task: единица работы, делегируемая от одного агента к другому. Имеет статус (submitted, working, completed, failed).

Artifact: результат выполнения задачи (текст, файл, структурированные данные).

Push Notifications: агент-сервер уведомляет клиента об изменении статуса через webhook.

Agent Card: описание возможностей агента

// GET /.well-known/agent.json
{
  "name": "Financial Analysis Agent",
  "description": "Агент финансового анализа корпоративных данных",
  "url": "https://fin-agent.company.com",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "authentication": {
    "schemes": ["Bearer"]
  },
  "skills": [
    {
      "id": "financial_analysis",
      "name": "Financial Analysis",
      "description": "Анализ финансовых данных, P&L, расчёт KPI",
      "inputModes": ["text"],
      "outputModes": ["text", "application/json"],
      "examples": [
        "Проанализируй выручку за Q1 2025 vs Q1 2024",
        "Рассчитай EBITDA по подразделениям"
      ]
    },
    {
      "id": "forecast",
      "name": "Revenue Forecast",
      "description": "Прогнозирование выручки на основе исторических данных",
      "inputModes": ["text", "application/json"],
      "outputModes": ["text", "application/json"]
    }
  ]
}

A2A-сервер на Python

# pip install a2a-sdk
from a2a.server.fastapi import A2AServer, A2ARouter
from a2a.types import AgentCard, AgentSkill, Task, TaskStatus, Artifact, TextArtifact
from fastapi import FastAPI
import uuid

# Agent Card
agent_card = AgentCard(
    name="Financial Analysis Agent",
    description="Агент финансового анализа",
    url="https://fin-agent.company.com",
    version="1.0.0",
    skills=[
        AgentSkill(
            id="financial_analysis",
            name="Financial Analysis",
            description="Анализ P&L, расчёт KPI, выявление аномалий",
        )
    ],
)

app = FastAPI()
a2a_router = A2ARouter(agent_card=agent_card)

@a2a_router.on_task("financial_analysis")
async def handle_financial_task(task: Task) -> Task:
    """Обработчик задач финансового анализа"""

    # Получаем входные данные
    user_input = task.input.message.parts[0].text

    # Выполняем анализ через внутреннего агента
    result = await financial_agent.analyze(user_input)

    # Возвращаем результат
    task.artifacts = [
        TextArtifact(
            name="analysis_result",
            parts=[{"type": "text", "text": result}],
        )
    ]
    task.status = TaskStatus(state="completed")

    return task

# Стриминг результата
@a2a_router.on_task("financial_analysis", streaming=True)
async def handle_streaming_task(task: Task):
    """Стриминговый обработчик"""
    user_input = task.input.message.parts[0].text

    async for chunk in financial_agent.stream_analyze(user_input):
        yield TaskStatus(state="working"), TextArtifact(
            name="partial_result",
            parts=[{"type": "text", "text": chunk}],
        )

app.include_router(a2a_router)

A2A-клиент: делегирование задач

from a2a.client import A2AClient

# Обнаружение агента
client = await A2AClient.from_url("https://fin-agent.company.com")
agent_card = client.agent_card

print(f"Агент: {agent_card.name}")
print(f"Навыки: {[s.name for s in agent_card.skills]}")

# Отправка задачи
task = await client.send_task(
    skill_id="financial_analysis",
    message="Проанализируй отклонение выручки от плана за март 2025",
)

# Ожидание результата
completed_task = await client.wait_for_completion(task.id)
print(completed_task.artifacts[0].parts[0]["text"])

# Стриминг
async for status, artifact in client.stream_task(
    skill_id="financial_analysis",
    message="Создай прогноз выручки на Q2 2025",
):
    if artifact:
        print(artifact.parts[0]["text"], end="", flush=True)

Интеграция с LangGraph

from langgraph.graph import StateGraph, END
from a2a.client import A2AClient

class OrchestratorState(TypedDict):
    task: str
    financial_result: Optional[str]
    legal_result: Optional[str]
    final_report: Optional[str]

# Нода, делегирующая задачу внешнему A2A-агенту
async def delegate_to_financial_agent(state: OrchestratorState):
    client = await A2AClient.from_url("https://fin-agent.company.com")
    task = await client.send_task(
        skill_id="financial_analysis",
        message=state["task"],
    )
    completed = await client.wait_for_completion(task.id, timeout=120)
    return {"financial_result": completed.artifacts[0].parts[0]["text"]}

async def delegate_to_legal_agent(state: OrchestratorState):
    client = await A2AClient.from_url("https://legal-agent.legalteam.com")
    task = await client.send_task(
        skill_id="contract_review",
        message=state["task"],
    )
    completed = await client.wait_for_completion(task.id, timeout=180)
    return {"legal_result": completed.artifacts[0].parts[0]["text"]}

# Оркестратор объединяет результаты от двух внешних агентов
graph = StateGraph(OrchestratorState)
graph.add_node("financial", delegate_to_financial_agent)
graph.add_node("legal", delegate_to_legal_agent)
graph.add_node("synthesize", synthesize_results)
# ...

Практический кейс: корпоративный маркетплейс агентов

Ситуация: холдинг с несколькими подразделениями, каждое разработало специализированных агентов (финансы, юристы, HR, логистика) на разных фреймворках.

A2A как интеграционный слой: оркестратор корпоративного портала делегирует задачи агентам через стандартный A2A-протокол, не зная о внутренней реализации каждого.

Схема:

• Portal Agent (LangGraph) → Financial Agent (Python/LlamaIndex) через A2A
• Portal Agent → Legal Agent (Node.js/LangChain) через A2A
• Portal Agent → HR Agent (Java/Semantic Kernel) через A2A

Результаты:

• Интеграция нового агента в экосистему: 1 день (публикация Agent Card + реализация A2A endpoint)
• Смена внутренней реализации агента: прозрачна для оркестратора
• Межкомандное сотрудничество: каждая команда владеет своим агентом

Текущий статус A2A

A2A — относительно новый стандарт (2025). Нативная поддержка в Google ADK, растущая поддержка в LangGraph и других фреймворках. При выборе между A2A и кастомным API оценивайте: есть ли несколько разных агентов от разных команд/провайдеров? Нужна ли стандартизированная discoverability? Если да — A2A оправдан.

Сроки

• Настройка A2A-сервера с агентом: 2–3 дня
• A2A-клиент в оркестраторе: 1–2 дня
• Полноценный маркетплейс с несколькими агентами: 2–3 недели