A2A Протокол от Google: Подключение к агентам

В эпоху, когда искусственный интеллект становится повсеместным, возникает критическая проблема: агенты, созданные на разных платформах, с разными фреймворками и библиотеками, не могут эффективно взаимодействовать друг с другом. Каждый агент — это как остров, изолированный от остального мира.

Google столкнулся с этой проблемой при масштабировании своих внутренних агентных систем. В апреле 2025 года компания представила Agent2Agent (A2A) Protocol — открытый стандарт для взаимодействия AI-агентов. Позже протокол был передан в Linux Foundation, став по-настоящему открытым проектом с сообществом разработчиков.

A2A можно сравнить с HTTP в ранних днях Интернета: это фундаментальный протокол, который позволяет различным системам общаться друг с другом без необходимости знать их внутреннее устройство.

Что Это Такое?

A2A (Agent-to-Agent) — это открытый протокол стандартизации взаимодействия между AI-агентами. Он определяет:

• Как агенты находят друг друга (discovery)
• Как они обмениваются сообщениями (communication)
• Как управляются задачи (task management)
• Как обеспечивается безопасность (authentication, authorization)
• Как обмениваются данными разных типов (text, files, structured data)

Ключевая идея A2A — opaque execution: агенты могут сотрудничать, не раскрывая своего внутреннего состояния, памяти, инструментов или логики. Они взаимодействуют исключительно через объявленные возможности и обмен информацией.

Кто Это Создал?

• Оригинальная разработка: Google
• Текущее владение: Linux Foundation
• Лицензия: Apache License 2.0
• GitHub репозиторий: https://github.com/a2aproject/A2A
• Официальная документация: https://a2a-protocol.org/latest/

Почему A2A Важен?

1. Интероперабельность: Агенты на LangGraph, CrewAI, Semantic Kernel и других фреймворках могут работать вместе
2. Комплексные сценарии: Агенты могут делегировать подзадачи друг другу, решая проблемы, недоступные одиночным агентам
3. Безопасность: Нет необходимости делиться внутренними данными и инструментами
4. Гибкость: Поддержка синхронных, асинхронных и потоковых взаимодействий
5. Enterprise-ready: Встроенные механизмы аутентификации, авторизации и мониторинга

Архитектура и Основные Концепции

A2A протокол имеет трехслойную архитектуру, что обеспечивает гибкость и расширяемость:

┌─────────────────────────────────────────────────────────────┐
│                     Layer 3: Protocol Bindings              │
│  • JSON-RPC Methods                                         │
│  • gRPC RPCs                                                │
│  • HTTP/REST Endpoints                                      │
│  • Custom Bindings                                          │
└─────────────────────────────────────────────────────────────┘
                            ↑
┌─────────────────────────────────────────────────────────────┐
│                     Layer 2: Abstract Operations            │
│  • Send Message                                             │
│  • Stream Message                                           │
│  • Get Task                                                 │
│  • List Tasks                                               │
│  • Cancel Task                                              │
│  • Subscribe to Task                                        │
│  • Push Notification Config                                 │
└─────────────────────────────────────────────────────────────┘
                            ↑
┌─────────────────────────────────────────────────────────────┐
│                     Layer 1: Canonical Data Model           │
│  • Task                                                     │
│  • Message                                                  │
│  • AgentCard                                                │
│  • Part                                                     │
│  • Artifact                                                 │
│  • Extension                                                │
└─────────────────────────────────────────────────────────────┘

Основные Концепции

A2A определяет восемь фундаментальных концепций:

1. A2A Client — приложение, инициирующее взаимодействие
2. A2A Server (Remote Agent) — агент, предоставляющий сервисы
3. Agent Card — метаданные агента для discovery
4. Message — единица коммуникации
5. Task — единица работы
6. Part — минимальная единица содержимого
7. Artifact — выходные данные
8. Extension — механизм расширения

Протокольные Привязки (Protocol Bindings)

A2A использует три основных протокольных привязки для реализации абстрактных операций:

1. JSON-RPC 2.0

JSON-RPC 2.0 — это протокол удаленного вызова процедур, использующий JSON для кодирования сообщений. A2A использует его как основной формат полезной нагрузки для всех запросов и ответов.

Структура JSON-RPC запроса:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "SendMessage",
  "params": { ... }
}

Структура JSON-RPC ответа:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": { ... }
}

JSON-RPC Method Names

В A2A протоколе JSON-RPC методы используют PascalCase без слэшей. Каждый метод соответствует gRPC RPC методу в прототипе.

2. HTTP/REST

A2A серверы предоставляют HTTP(S) endpoint для обработки JSON-RPC запросов.

Основные HTTP методы:

• POST /a2a/v1 — отправка JSON-RPC запросов
• GET /.well-known/agent-card.json — получение Agent Card

3. Server-Sent Events (SSE)

Для потоковой передачи A2A использует Server-Sent Events (SSE) — стандарт W3C для односторонней потоковой передачи данных от сервера к клиенту.

Пример заголовков SSE:

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
Transfer-Encoding: chunked

Обзор API

Agent Discovery: Получение Agent Card

Технические детали приведенные в статье могут стать не актуальными в связи с изменениями версий протокола - он живой и активно развивается

Перед взаимодействием с агентом клиент должен получить его Agent Card.

HTTP запрос:

GET /.well-known/agent-card.json HTTP/1.1
Host: agent.example.com

Пример ответа

{
  "name": "Math Agent",
  "description": "Performs mathematical calculations",
  "url": "https://agent.example.com/a2a/v1",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "securitySchemes": {
    "apiKey": {
      "type": "apiKey",
      "in": "header",
      "name": "X-API-Key"
    }
  },
  "protocolVersions": ["1.0"]
}

Python SDK: Практическое Руководство

A2A имеет официальный Python SDK, который упрощает создание клиентов и серверов.

Установка

pip install a2a-sdk

Создание A2A Клиента

from a2a.client import A2AClient
from a2a.types import SendMessageRequest, Message, Part, TextPart
from uuid import uuid4

# Создаем клиента
client = A2AClient(
    endpoint_url="https://agent.example.com/a2a/v1",
    api_key="your-api-key"  # Если требуется аутентификация
)

# Создаем сообщение
message = Message(
    role="user",
    parts=[TextPart(text="Calculate the factorial of 10")],
    message_id=str(uuid4())
)

# Отправляем сообщение
request = SendMessageRequest(message=message)
response = client.send_message(request)

print(f"Task ID: {response.task.id}")
print(f"Task status: {response.task.status.state}")

Создание A2A Сервера

from a2a.server import A2AServer
from a2a.types import Task, TaskStatus, TaskState, Artifact, Part, TextPart
from uuid import uuid4
import asyncio

# Создаем простой агент, который отвечает на математические вопросы
class MathAgent:
    async def handle_message(self, message):
        text = message.parts[0].text

        # Простая логика для вычисления факториала
        if "factorial" in text.lower():
            # Извлекаем число из текста
            import re
            numbers = re.findall(r'\d+', text)
            if numbers:
                n = int(numbers[0])
                result = 1
                for i in range(1, n + 1):
                    result *= i

                artifact = Artifact(
                    artifact_id=str(uuid4()),
                    name="factorial_result",
                    parts=[TextPart(text=f"The factorial of {n} is {result}")]
                )

                return Task(
                    id=str(uuid4()),
                    context_id=str(uuid4()),
                    status=TaskStatus(state=TaskState.completed),
                    artifacts=[artifact]
                )

        # Если не распознано, возвращаем стандартный ответ
        artifact = Artifact(
            artifact_id=str(uuid4()),
            name="response",
            parts=[TextPart(text="I can calculate factorials. Try asking: 'What is the factorial of 5?'")]
        )

        return Task(
            id=str(uuid4()),
            context_id=str(uuid4()),
            status=TaskStatus(state=TaskState.completed),
            artifacts=[artifact]
        )

# Создаем сервер
server = A2AServer(
    host="0.0.0.0",
    port=8000,
    agent=MathAgent()
)

# Запускаем сервер
asyncio.run(server.start())

Потоковая Передача (Streaming)

from a2a.client import A2AClient
from a2a.types import SendMessageRequest, Message, TextPart
from uuid import uuid4

client = A2AClient(endpoint_url="https://agent.example.com/a2a/v1")

message = Message(
    role="user",
    parts=[TextPart(text="Write a long essay about AI")],
    message_id=str(uuid4())
)

request = SendMessageRequest(message=message)

# Используем потоковую передачу
for event in client.stream_message(request):
    if hasattr(event, 'task'):
        print(f"Task started: {event.task.id}")
    elif hasattr(event, 'artifact'):
        print(f"Artifact update: {event.artifact.parts[0].text[:50]}...")
    elif hasattr(event, 'status'):
        print(f"Status: {event.status.state}")

Обработка Push Уведомлений

from fastapi import FastAPI, Request, HTTPException
from a2a.types import Task
import json

app = FastAPI()

@app.post("/webhook/a2a-notifications")
async def handle_a2a_notification(request: Request):
    # Получаем токен из заголовка
    notification_token = request.headers.get("X-A2A-Notification-Token")

    # Валидация токена (в реальном приложении проверяйте подписи и т.д.)
    if not notification_token:
        raise HTTPException(status_code=401, detail="Missing notification token")

    # Получаем тело запроса
    body = await request.json()

    # Парсим Task из JSON
    task = Task(**body)

    # Обрабатываем уведомление
    if task.status.state == "completed":
        print(f"Task {task.id} completed successfully!")
        # Здесь можно отправить уведомление пользователю, обновить БД и т.д.
    elif task.status.state == "failed":
        print(f"Task {task.id} failed: {task.status.message}")

    return {"status": "ok"}

Примеры JSON-RPC Запросов и Ответов

1. Отправка Сообщения (SendMessage)

Запрос:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "Calculate 15 * 27"
        }
      ],
      "messageId": "9229e770-767c-417b-a0b0-f0741243c589"
    },
    "metadata": {}
  }
}

Ответ (Task-based):

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "id": "363422be-b0f9-4692-a24d-278670e7c7f1",
    "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4",
    "status": {
      "state": "completed",
      "timestamp": "2025-04-02T16:59:35.331Z"
    },
    "artifacts": [
      {
        "artifactId": "9b6934dd-37e3-4eb1-8766-962efaab63a1",
        "name": "calculation",
        "parts": [
          {
            "kind": "text",
            "text": "15 * 27 = 405"
          }
        ]
      }
    ],
    "history": [
      {
        "role": "user",
        "parts": [
          {
            "kind": "text",
            "text": "Calculate 15 * 27"
          }
        ],
        "messageId": "9229e770-767c-417b-a0b0-f0741243c589",
        "taskId": "363422be-b0f9-4692-a24d-278670e7c7f1",
        "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4"
      }
    ],
    "kind": "task",
    "metadata": {}
  }
}

Коды ошибок JSON-RPC 2.0:

2. Получение Статуса Задачи (GetTask)

Запрос:

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "GetTask",
  "params": {
    "id": "363422be-b0f9-4692-a24d-278670e7c7f1"
  }
}

Ответ:

{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "id": "363422be-b0f9-4692-a24d-278670e7c7f1",
    "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4",
    "status": {
      "state": "completed",
      "timestamp": "2025-04-02T16:59:35.331Z"
    },
    "artifacts": [
      {
        "artifactId": "9b6934dd-37e3-4eb1-8766-962efaab63a1",
        "name": "calculation",
        "parts": [
          {
            "kind": "text",
            "text": "15 * 27 = 405"
          }
        ]
      }
    ],
    "kind": "task"
  }
}

3. Потоковая Передача (SendStreamingMessage)

Запрос:

{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "SendStreamingMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "Write a poem about Python"
        }
      ],
      "messageId": "bbb7dee1-cf5c-4683-8a6f-4114529da5eb"
    },
    "metadata": {}
  }
}

SSE Response (поток событий):

data: {"jsonrpc": "2.0", "id": 3, "result": {"id": "task-123", "status": {"state": "submitted"}, "kind": "task"}}

data: {"jsonrpc": "2.0", "id": 3, "result": {"taskId": "task-123", "artifact": {"parts": [{"text": "In the realm of code"}], "append": false, "kind": "artifact-update"}}}

data: {"jsonrpc": "2.0", "id": 3, "result": {"taskId": "task-123", "artifact": {"parts": [{"text": "Where Python flows"}], "append": true, "kind": "artifact-update"}}}

data: {"jsonrpc": "2.0", "id": 3, "result": {"taskId": "task-123", "status": {"state": "completed"}, "final": true, "kind": "status-update"}}

4. Многоходовое Взаимодействие

Шаг 1: Инициация

{
  "jsonrpc": "2.0",
  "id": 4,
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"kind": "text", "text": "Book a flight"}],
      "messageId": "c53ba666-3f97-433c-a87b-6084276babe2"
    }
  }
}

Ответ (input-required):

{
  "jsonrpc": "2.0",
  "id": 4,
  "result": {
    "id": "3f36680c-7f37-4a5f-945e-d78981fafd36",
    "status": {
      "state": "input-required",
      "message": {
        "role": "agent",
        "parts": [{"kind": "text", "text": "Where to, where from, and when?"}]
      }
    },
    "kind": "task"
  }
}

Шаг 2: Ответ на запрос

{
  "jsonrpc": "2.0",
  "id": 5,
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"kind": "text", "text": "NYC to London, October 10"}],
      "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4",
      "taskId": "3f36680c-7f37-4a5f-945e-d78981fafd36",
      "messageId": "0db1d6c4-3976-40ed-b9b8-0043ea7a03d3"
    }
  }
}

5. Обмен Файлами

Загрузка изображения:

{
  "jsonrpc": "2.0",
  "id": 6,
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {"kind": "text", "text": "Analyze this image"},
        {
          "kind": "file",
          "file": {
            "name": "input.png",
            "mimeType": "image/png",
            "bytes": "iVBORw0KGgoAAAANSUhEUgAAAAUA..."
          }
        }
      ],
      "messageId": "file-upload-123"
    }
  }
}

Получение обработанного файла:

{
  "jsonrpc": "2.0",
  "id": 6,
  "result": {
    "id": "file-task-123",
    "artifacts": [
      {
        "artifactId": "processed-img-123",
        "parts": [
          {
            "kind": "file",
            "file": {
              "name": "output.png",
              "mimeType": "image/png",
              "uri": "https://storage.example.com/processed/task-123/output.png?token=xyz"
            }
          }
        ]
      }
    ],
    "kind": "task"
  }
}

6. Push Уведомления

Настройка push уведомления:

{
  "jsonrpc": "2.0",
  "id": 7,
  "method": "SendMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [{"kind": "text", "text": "Generate Q1 report"}],
      "messageId": "report-123"
    },
    "configuration": {
      "pushNotificationConfig": {
        "url": "https://client.example.com/webhook/a2a-notifications",
        "token": "secure-client-token",
        "authentication": {"schemes": ["Bearer"]}
      }
    }
  }
}

Push уведомление (HTTP POST на webhook):

POST /webhook/a2a-notifications HTTP/1.1
Host: client.example.com
Content-Type: application/json
X-A2A-Notification-Token: secure-client-token

{
  "id": "report-task-123",
  "status": {"state": "completed"},
  "kind": "task"
}

Жизненный Цикл Задачи

Задача в A2A проходит через следующие состояния:

┌──────────────────────┐     ┌──────────────────┐     ┌───────────────────────┐
│ TASK_STATE_SUBMITTED │────▶│ TASK_STATE_WORKING │────▶│ TASK_STATE_COMPLETED  │
└──────────────────────┘     └──────────────────┘     └───────────────────────┘
         ▲                           │                          │
         │                           ▼                          ▼
         │                    ┌───────────────────────┐     ┌──────────────────┐
         │                    │TASK_STATE_INPUT_REQUIRED│     │TASK_STATE_FAILED │
         │                    └───────────────────────┘     └──────────────────┘
         │                           │
         │                           ▼
         │                    ┌──────────────────┐
         └───────────────────│TASK_STATE_CANCELED │
                             └──────────────────┘
         ▲
         │
         │                    ┌──────────────────┐
         └───────────────────│TASK_STATE_REJECTED │
                             └──────────────────┘
         ▲
         │
         │                    ┌──────────────────────┐
         └───────────────────│TASK_STATE_AUTH_REQUIRED│
                             └──────────────────────┘

Task States (Состояния Задачи)

A2A v1.0.0 определяет 9 состояний задачи (TaskState enum):

Получение Списка Задач (ListTasks)

Запрос:

{
  "jsonrpc": "2.0",
  "id": 8,
  "method": "ListTasks",
  "params": {
    "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4",
    "status": "working",
    "pageSize": 10,
    "includeArtifacts": false
  }
}

Ответ:

{
  "jsonrpc": "2.0",
  "id": 8,
  "result": {
    "tasks": [
      {
        "id": "task-1",
        "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4",
        "status": {"state": "working"},
        "kind": "task"
      },
      {
        "id": "task-2",
        "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4",
        "status": {"state": "working"},
        "kind": "task"
      }
    ],
    "nextPageToken": "next-token-123",
    "pageSize": 10,
    "totalSize": 2
  }
}

Аутентификация и Безопасность

A2A использует стандартные веб-механизмы безопасности:

1. Transport Security

• Обязательно: HTTPS для production
• Рекомендуется: TLS 1.2+

2. Аутентификация через Agent Card

Agent Card содержит информацию о требованиях аутентификации:

{
  "securitySchemes": {
    "google": {
      "type": "openIdConnect",
      "openIdConnectUrl": "https://accounts.google.com/.well-known/openid-configuration"
    },
    "apiKey": {
      "type": "apiKey",
      "in": "header",
      "name": "X-API-Key"
    }
  },
  "security": [{"google": ["openid", "profile", "email"]}]
}

3. Типы Аутентификации

API Key:

POST /a2a/v1 HTTP/1.1
Host: agent.example.com
Content-Type: application/json
X-API-Key: your-api-key

{...}

Bearer Token (OAuth 2.0):

POST /a2a/v1 HTTP/1.1
Host: agent.example.com
Content-Type: application/json
Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

{...}

Basic Auth:

POST /a2a/v1 HTTP/1.1
Host: agent.example.com
Content-Type: application/json
Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=

{...}

4. Валидация Webhook

При использовании push уведомлений, webhook должен валидировать запросы:

import hmac
import hashlib
from fastapi import Request, HTTPException

async def validate_webhook(request: Request, secret: str):
    # Получаем подпись из заголовка
    signature = request.headers.get("X-A2A-Signature")

    if not signature:
        raise HTTPException(status_code=401, detail="Missing signature")

    # Получаем тело запроса
    body = await request.body()

    # Вычисляем HMAC-SHA256
    expected_signature = hmac.new(
        secret.encode(),
        body,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected_signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    return True

Агент Карточка Структура (Agent Card Structure)

Agent Card — это метаданные агента для discovery. Обязательные поля:

Пример Agent Card

{
  "name": "Math Assistant",
  "description": "Агент для математических вычислений",
  "supportedInterfaces": ["a2a-json-rpc", "a2a-grpc"],
  "provider": "Example Corp",
  "version": "1.0.0",
  "documentationUrl": "https://example.com/docs/math-agent",
  "capabilities": ["calculation", "analysis"],
  "securitySchemes": {
    "apiKey": {
      "type": "apiKey",
      "in": "header",
      "name": "X-API-Key"
    }
  },
  "securityRequirements": ["apiKey"],
  "defaultInputModes": ["text", "file"],
  "defaultOutputModes": ["text"],
  "skills": ["arithmetic", "algebra", "calculus"],
  "signatures": [
    {
      "algorithm": "ES256",
      "keyId": "key-1",
      "signature": "base64-encoded-signature"
    }
  ]
}

Обработка Ошибок (Error Handling)

A2A определяет 9 специфичных кодов ошибок:

Формат Времени (Timestamp Format)

Все временные метки в A2A должны использовать ISO 8601 формат с UTC timezone:

YYYY-MM-DDTHH:mm:ss.sssZ

Пример: "2025-04-02T16:59:35.331Z"

Это обеспечивает однозначную интерпретацию времени независимо от часовых поясов.

Версия Протокола (Protocol Version)

Текущая версия A2A протокола: v1.0.0 (выпущена в октябре 2025).

Все реализации должны указывать версию протокола, которую они поддерживают.

Соглашение об Именах Полей JSON (JSON Field Naming Convention)

Все JSON поля в A2A должны использовать camelCase:

• Protocol Buffer protocol_version → JSON protocolVersion
• Protocol Buffer context_id → JSON contextId
• Protocol Buffer default_input_modes → JSON defaultInputModes
• Protocol Buffer security_requirements → JSON securityRequirements

Это обеспечивает согласованность и совместимость с JavaScript/TypeScript экосистемой.

Формат Значений Перечислений (Enum Values Format)

Все enum значения в JSON должны использовать SCREAMING_SNAKE_CASE:

• TASK_STATE_COMPLETED (не completed)
• TASK_STATE_AUTH_REQUIRED (не auth-required)
• TASK_STATE_UNSPECIFIED (не unspecified)

Это соответствует ProtoJSON спецификации и обеспечивает однозначную интерпретацию.

gRPC Protocol Binding (Привязка gRPC Протокола)

A2A также определяет gRPC binding для прямого использования gRPC:

Endpoint: a2a.A2A

RPC Methods:

• SendMessage(SendMessageRequest) returns (SendMessageResponse)
• SendStreamingMessage(SendMessageRequest) returns (stream SendMessageResponse)
• GetTask(GetTaskRequest) returns (GetTaskResponse)
• ListTasks(ListTasksRequest) returns (ListTasksResponse)
• CancelTask(CancelTaskRequest) returns (CancelTaskResponse)
• SubscribeToTask(SubscribeToTaskRequest) returns (stream SubscribeToTaskResponse)
• CreateTaskPushNotificationConfig(TaskPushNotificationConfig) returns (TaskPushNotificationConfig)
• GetTaskPushNotificationConfig(GetTaskPushNotificationConfigRequest) returns (TaskPushNotificationConfig)
• ListTaskPushNotificationConfigs(ListTaskPushNotificationConfigsRequest) returns (ListTaskPushNotificationConfigsResponse)
• DeleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigRequest) returns (Empty)
• GetExtendedAgentCard(GetExtendedAgentCardRequest) returns (GetExtendedAgentCardResponse)

Service Definition:

service A2A {
  rpc SendMessage(SendMessageRequest) returns (SendMessageResponse);
  rpc SendStreamingMessage(SendMessageRequest) returns (stream SendMessageResponse);
  rpc GetTask(GetTaskRequest) returns (GetTaskResponse);
  rpc ListTasks(ListTasksRequest) returns (ListTasksResponse);
  rpc CancelTask(CancelTaskRequest) returns (CancelTaskResponse);
  rpc SubscribeToTask(SubscribeToTaskRequest) returns (stream SubscribeToTaskResponse);
  rpc CreateTaskPushNotificationConfig(TaskPushNotificationConfig) returns (TaskPushNotificationConfig);
  rpc GetTaskPushNotificationConfig(GetTaskPushNotificationConfigRequest) returns (TaskPushNotificationConfig);
  rpc ListTaskPushNotificationConfigs(ListTaskPushNotificationConfigsRequest) returns (ListTaskPushNotificationConfigsResponse);
  rpc DeleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigRequest) returns (Empty);
  rpc GetExtendedAgentCard(GetExtendedAgentCardRequest) returns (GetExtendedAgentCardResponse);
}

Практические Примеры

Пример 1: Многоагентная Система

Создадим систему из трех агентов, которые работают вместе:

# agent1.py - Агент анализа данных
from a2a.server import A2AServer
from a2a.types import Task, TaskStatus, TaskState, Artifact, TextPart
from uuid import uuid4

class DataAnalysisAgent:
    async def handle_message(self, message):
        text = message.parts[0].text

        # Имитация анализа данных
        artifact = Artifact(
            artifact_id=str(uuid4()),
            name="analysis_report",
            parts=[TextPart(text=f"Data analysis complete. Found {len(text)} characters.")]
        )

        return Task(
            id=str(uuid4()),
            context_id=str(uuid4()),
            status=TaskStatus(state=TaskState.completed),
            artifacts=[artifact]
        )

# agent2.py - Агент генерации отчетов
class ReportGenerationAgent:
    async def handle_message(self, message):
        text = message.parts[0].text

        # Генерация отчета
        artifact = Artifact(
            artifact_id=str(uuid4()),
            name="generated_report",
            parts=[TextPart(text=f"Report generated based on: {text}")]
        )

        return Task(
            id=str(uuid4()),
            context_id=str(uuid4()),
            status=TaskStatus(state=TaskState.completed),
            artifacts=[artifact]
        )

# agent3.py - Оркестратор
class OrchestratorAgent:
    def __init__(self):
        self.client = A2AClient(endpoint_url="http://localhost:8001/a2a/v1")

    async def handle_message(self, message):
        # Отправляем данные на анализ
        analysis_task = self.client.send_message(
            SendMessageRequest(message=message)
        )

        # Получаем результат анализа
        analysis_result = self.client.get_task(analysis_task.task.id)

        # Отправляем результат на генерацию отчета
        report_message = Message(
            role="user",
            parts=[TextPart(text=analysis_result.task.artifacts[0].parts[0].text)],
            message_id=str(uuid4())
        )

        report_task = self.client.send_message(
            SendMessageRequest(message=report_message)
        )

        return report_task.task

Пример 2: Агент с Файловым Обменом

import base64
from a2a.client import A2AClient
from a2a.types import SendMessageRequest, Message, Part, TextPart, FilePart
from uuid import uuid4

client = A2AClient(endpoint_url="https://image-agent.example.com/a2a/v1")

# Загружаем изображение
with open("input.jpg", "rb") as f:
    image_data = base64.b64encode(f.read()).decode("utf-8")

message = Message(
    role="user",
    parts=[
        TextPart(text="Analyze this image and describe what you see"),
        FilePart(
            name="input.jpg",
            mime_type="image/jpeg",
            data=image_data
        )
    ],
    message_id=str(uuid4())
)

request = SendMessageRequest(message=message)
response = client.send_message(request)

print(f"Task ID: {response.task.id}")

# Получаем результат
task = client.get_task(response.task.id)
for artifact in task.task.artifacts:
    for part in artifact.parts:
        if hasattr(part, 'file') and part.file.uri:
            print(f"Processed image available at: {part.file.uri}")

Пример 3: Агент с Многоходовым Взаимодействием

from a2a.client import A2AClient
from a2a.types import SendMessageRequest, Message, TextPart
from uuid import uuid4

client = A2AClient(endpoint_url="https://booking-agent.example.com/a2a/v1")

# Шаг 1: Инициация
message1 = Message(
    role="user",
    parts=[TextPart(text="Book a hotel")],
    message_id=str(uuid4())
)

response1 = client.send_message(
    SendMessageRequest(message=message1)
)

task_id = response1.task.id
context_id = response1.task.context_id

print(f"Task started: {task_id}")

# Шаг 2: Ответ на запрос
message2 = Message(
    role="user",
    parts=[TextPart(text="New York, July 15-20")],
    message_id=str(uuid4()),
    context_id=context_id,
    task_id=task_id
)

response2 = client.send_message(
    SendMessageRequest(message=message2)
)

print(f"Task status: {response2.task.status.state}")

# Шаг 3: Подтверждение
message3 = Message(
    role="user",
    parts=[TextPart(text="Confirm the first option")],
    message_id=str(uuid4()),
    context_id=context_id,
    task_id=task_id
)

response3 = client.send_message(
    SendMessageRequest(message=message3)
)

print(f"Booking confirmed: {response3.task.status.state == 'completed'}")

Миграция с Других Протоколов

Из MCP (Model Context Protocol)

MCP и A2A дополняют друг друга:

• MCP: агент ↔ инструменты (tools, APIs, resources)
• A2A: агент ↔ агент (agent-to-agent communication)

Пример интеграции:

# Агент использует MCP для доступа к инструментам
from mcp import MCPClient

# Агент использует A2A для взаимодействия с другими агентами
from a2a.client import A2AClient

class IntegratedAgent:
    def __init__(self):
        self.mcp_client = MCPClient(url="http://localhost:3001")
        self.a2a_client = A2AClient(endpoint_url="http://localhost:8002/a2a/v1")

    async def handle_a2a_request(self, message):
        # Используем MCP инструменты для обработки запроса
        tools_result = await self.mcp_client.call_tool("search_web", {"query": message.parts[0].text})

        # Отправляем результат другому агенту через A2A
        result_message = Message(
            role="agent",
            parts=[TextPart(text=tools_result)],
            message_id=str(uuid4())
        )

        response = self.a2a_client.send_message(
            SendMessageRequest(message=result_message)
        )

        return response.task

Лучшие Практики

1. Обработка Ошибок

from a2a.types import A2AError, TaskNotFoundError, UnsupportedOperationError

try:
    response = client.send_message(request)
except TaskNotFoundError as e:
    print(f"Task not found: {e}")
except UnsupportedOperationError as e:
    print(f"Operation not supported: {e}")
except A2AError as e:
    print(f"A2A error: {e}")

2. Управление Сессиями

# Используйте contextId для группировки связанных задач
context_id = str(uuid4())

# Все сообщения в одной сессии используют один contextId
message1 = Message(
    role="user",
    parts=[TextPart(text="Start project")],
    message_id=str(uuid4()),
    context_id=context_id
)

message2 = Message(
    role="user",
    parts=[TextPart(text="Add task")],
    message_id=str(uuid4()),
    context_id=context_id,
    task_id=previous_task_id
)

3. Потоковая Передача для Долгих Операций

# Используйте streaming для операций, которые занимают больше 5 секунд
for event in client.stream_message(request):
    if hasattr(event, 'artifact'):
        # Обновляем UI с промежуточными результатами
        update_ui(event.artifact.parts[0].text)

4. Валидация Agent Card

def validate_agent_card(card):
    required_fields = ["name", "description", "supportedInterfaces", "provider", "version", "capabilities", "securitySchemes", "securityRequirements", "defaultInputModes", "defaultOutputModes", "skills", "signatures"]

    for field in required_fields:
        if field not in card:
            raise ValueError(f"Missing required field: {field}")

    if not card["supportedInterfaces"]:
        raise ValueError("Agent must have supportedInterfaces defined")

    if not card["capabilities"]:
        raise ValueError("Agent must have capabilities defined")

    return True

Заключение

A2A протокол v1.0.0 — это фундаментальный шаг в развитии экосистемы AI-агентов. Он решает проблему фрагментации, позволяя агентам, созданным на разных платформах, эффективно взаимодействовать друг с другом.

Ключевые Преимущества

1. Интероперабельность: Работа между агентами на разных фреймворках
2. Безопасность: Opaque execution без раскрытия внутреннего состояния
3. Гибкость: Поддержка синхронных, асинхронных и потоковых взаимодействий
4. Enterprise-ready: Встроенные механизмы аутентификации и авторизации
5. Open Standard: Разрабатывается под эгидой Linux Foundation

Когда Использовать A2A

• Создание многоагентных систем
• Интеграция агентов на разных платформах
• Построение микросервисной архитектуры для AI
• Enterprise-приложения с требованиями безопасности

Дальнейшее Изучение

• Официальная документация: https://a2a-protocol.org/latest/
• GitHub репозиторий: https://github.com/a2aproject/A2A
• Python SDK: https://pypi.org/project/a2a-sdk/
• Примеры: https://github.com/a2aproject/A2A/tree/main/examples

Справочник

JSON-RPC Методы

Типы Part

Состояния Задачи

Ссылки и Источники

Официальная Документация и Репозитории

1. A2A Protocol Documentation. a2a-protocol.org. URL: https://a2a-protocol.org/latest/ (дата обращения: 2026-04-08)

2. A2A GitHub Repository. GitHub. URL: https://github.com/a2aproject/A2A (дата обращения: 2026-04-08)

3. A2A Python SDK on PyPI. Python Package Index. URL: https://pypi.org/project/a2a-sdk/ (дата обращения: 2026-04-08)

Смежные Протоколы и Стандарты

4. Model Context Protocol (MCP). Anthropic. URL: https://modelcontextprotocol.org/ (дата обращения: 2026-04-08)

5. JSON-RPC 2.0 Specification. jsonrpc.org. URL: https://www.jsonrpc.org/specification (дата обращения: 2026-04-08)

6. Server-Sent Events (SSE) Specification. WHATWG. URL: https://html.spec.whatwg.org/multipage/server-sent-events.html (дата обращения: 2026-04-08)

7. HTTP/REST Best Practices. MDN Web Docs. URL: https://developer.mozilla.org/en-US/docs/Web/HTTP/Overview (дата обращения: 2026-04-08)

8. OAuth 2.0 Authorization Framework. IETF. URL: https://datatracker.ietf.org/doc/html/rfc6749 (дата обращения: 2026-04-08)

Лицензия

9. Apache License 2.0. Apache Software Foundation. URL: https://www.apache.org/licenses/LICENSE-2.0 (дата обращения: 2026-04-08)

Исторические Материалы

10. Google A2A Announcement. Google Developers Blog. URL: https://developers.googleblog.com/2025/04/agent-to-agent-protocol.html (9 апреля 2025) (дата обращения: 2026-04-08)

11. Linux Foundation A2A Donation. LF AI & Data Blog. URL: https://lfaidata.foundation/blog/2025/06/23/a2a-joins-lf-ai-data/ (23 июня 2025) (дата обращения: 2026-04-08)

Дополнительные Ресурсы

12. A2A Protocol Specification. a2a-protocol.org. URL: https://a2a-protocol.org/latest/specification.html (дата обращения: 2026-04-08)

13. A2A Python SDK Examples. GitHub. URL: https://github.com/a2aproject/A2A/tree/main/examples (дата обращения: 2026-04-08)