Начало работы с протоколом A2A

Протокол агентного-агент (A2A) является открытым стандартом, инициированным Google, предназначенным для обеспечения связи и совместимости между различными системами AI-агентов. Этот учебник поможет вам освоить основы протокола и настроить ваше первое агентное общение.

Что такое протокол A2A?

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

• Узнавать о возможностях друг друга's
• Договариваться о режимах взаимодействия (текст, формы, файлы)
• Сотрудничать над задачами безопасно
• Обмениваться структурированными данными и артефактами

Протокол использует JSON-RPC 2.0 через HTTP(S) и поддерживает как стандартные схемы запрос/ответ, так и потоковую передачу через Server-Sent Events (SSE).

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

Прежде чем приступать к реализации, давайте разберемся с ключевыми концепциями:

• Карта агента: Публичный метаданный файл (обычно по адресу /.well-known/agent.json), описывающий возможности, навыки, URL-адрес конечной точки и требования к аутентификации агента.
• A2A сервер: Агент, предоставляющий HTTP конечную точку, реализующую методы протокола A2A. Он принимает запросы и управляет выполнением задач.
• A2A клиент: Приложение или другой агент, использующий услуги A2A. Он отправляет запросы на URL-адрес A2A сервера.
• Задача: Центральная единица работы. Клиент инициирует задачу, отправив сообщение. Задачи имеют уникальные идентификаторы и проходят через состояния.
• Сообщение: Представляет повороты общения между клиентом и агентом. Сообщения содержат Части.
• Часть: Основная единица содержания в Сообщении или Артефакте. Может быть TextPart, FilePart или DataPart.
• Артефакт: Представляет выходные данные, созданные агентом в ходе задачи (например, созданные файлы, структурированные данные).

Предварительные требования

Перед началом убедитесь, что у вас:

• Установлен Python 3.12+ или Node.js 18+
• Базовые знания HTTP API и JSON
• Знакомство с концепциями асинхронного программирования

Настройка вашей среды

Давайте настроим простой пример клиент-сервер A2A. Мы будем использовать Python для этого урока, но аналогичные концепции применимы к JavaScript.

1. Установите необходимые пакеты

# Создайте виртуальную среду (рекомендуется)
python -m venv a2a-env
source a2a-env/bin/activate  # На Windows: a2a-env\\Scripts\\activate

# Установите зависимости
pip install httpx pydantic

2. Создайте простой A2A клиент

Создайте файл a2a_client.py:

import httpx
import json
import uuid
from typing import Dict, Any, Optional

class A2AClient:
    def __init__(self, server_url: str):
        # Убедитесь, что URL не заканчивается на слэш
        self.url = server_url.rstrip("/")

    async def send_task(self, message: str) -> Dict[str, Any]:
        """Отправить простую задачу на A2A сервер."""
        task_id = str(uuid.uuid4())

        # Подготовьте JSON-RPC запрос
        payload = {
            "id": task_id,
            "message": {
                "role": "user",
                "parts": [{"text": message}]
            }
        }

        # Отправьте запрос
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.url}/tasks/send",
                json={
                    "jsonrpc": "2.0",
                    "method": "tasks/send",
                    "params": payload,
                    "id": 1
                }
            )
            response.raise_for_status()
            result = response.json()

            # Верните результат
            return result.get("result", {})

    async def get_task(self, task_id: str) -> Optional[Dict[str, Any]]:
        """Получить статус задачи."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.url}/tasks/get",
                json={
                    "jsonrpc": "2.0",
                    "method": "tasks/get",
                    "params": {"id": task_id},
                    "id": 1
                }
            )
            response.raise_for_status()
            result = response.json()

            # Верните результат
            return result.get("result", {})

3. Создайте простой тестовый скрипт

Создайте файл test_a2a.py:

import asyncio
from a2a_client import A2AClient

async def main():
    # Замените на URL вашего A2A сервера
    client = A2AClient("http://localhost:10000")

    try:
        # Отправьте задачу
        print("Отправка задачи на A2A сервер...")
        task = await client.send_task("Привет, A2A сервер! Что ты можешь сделать?")

        print(f"Task ID: {task.get('id')}")
        print(f"Task status: {task.get('status', {}).get('state')}")

        # Если задача имеет ответное сообщение
        if task.get('messages') and len(task['messages']) > 1:
            agent_message = task['messages'][1]  # Ответ агента
            if agent_message.get('parts'):
                for part in agent_message['parts']:
                    if part.get('text'):
                        print(f"Agent response: {part['text']}")

    except Exception as e:
        print(f"Error: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Обычный поток

Обычный поток коммуникации A2A следует следующим шагам:

1. Обнаружение: Клиент извлекает карту агента с общеизвестного URL сервера.
2. Инициация: Клиент отправляет запрос tasks/send или tasks/sendSubscribe, содержащий начальное пользовательское сообщение и уникальный идентификатор задачи.
3. Обработка: * (Потоковая передача): Сервер отправляет события SSE (обновления статуса, артефакты) по мере выполнения задачи. * (Не потоковая): Сервер синхронно обрабатывает задачу и возвращает окончательный объект Task в ответе.
4. Взаимодействие (Необязательно): Если задача переходит в состояние input-required, клиент отправляет последующие сообщения, используя тот же идентификатор задачи.
5. Завершение: Задача в конечном итоге достигает конечного состояния (completed, failed, canceled).

Запуск примера

Чтобы запустить этот пример, вам нужен A2A сервер. Вы можете использовать одного из образцов агентов из репозитория A2A:

1. Клонируйте репозиторий A2A:

git clone https://github.com/google-a2a/a2a-samples
cd a2a-samples

2. Выполните инструкции для запуска одного из образцов агентов, например, LangGraph Currency Agent:

cd samples/python/agents/langgraph
# Следуйте инструкциям по настройке в README

3. Как только агент будет запущен, выполните ваш тестовый скрипт:

python test_a2a.py

Следующие шаги

Теперь, когда вы понимаете основы протокола A2A, вы можете:

1. Изучить спецификацию протокола A2A для понимания всех возможностей.
2. Испытать различные образцы агентов из репозитория A2A.
3. Реализовать потоковую передачу с tasks/sendSubscribe для обновлений в реальном времени.
4. Узнать, как создать свой собственный сервер A2A для предоставления возможностей вашего агента.
5. Перейти к учебнику "Реализация A2A в вашем приложении" для более продвинутых шаблонов использования.
6. Узнайте, как использовать A2A Python SDK

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