ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать протокол A2A с discovery и capability negotiation

1. Цель задачи

Разработать и реализовать протокол взаимодействия между автономными AI-агентами (Agent-to-Agent, A2A), включающий механизмы обнаружения (discovery) и согласования возможностей (capability negotiation). В результате агенты должны уметь находить друг друга по описанию своих способностей и устанавливать канал для дальнейшего обмена сообщениями или выполнения кооперативных задач.

Ключевой результат Демонстрация, в которой два (или более) агента динамически находят друг друга в сети на основе заявленных capabilities и успешно завершают handshake с передачей согласованных параметров взаимодействия.

2. Исходные данные

Если нет реального инструмента — симулируем:

1. Если нет реального реестра, создаём простой HTTP-сервер на FastAPI, который хранит список агентов с их capabilities.
2. Вместо mDNS используем HTTP-запросы к реестру (централизованная модель discovery).
3. Если не хватает агентов, пишем два демо-агента с разными capabilities (например: translate, summarize, code_review).

3. Технологический стек

4. Этапы выполнения

Этап 1: Проектирование протокола A2A (1 час)

Действия

1. Определить структуру сообщений на основе JSON Schema.

   • AgentManifest – описание агента: agent_id, name, endpoint, capabilities (список строк или мапа строк).
   • DiscoveryRequest – { "required_capabilities": [...] }.
   • DiscoveryResponse – список подходящих агентов.
   • NegotiationRequest – { "session_id": "...", "capabilities": [...], "parameters": {...} }.
   • NegotiationResponse – { "session_id": "...", "accepted": true/false, "parameters": {...} }.

2. Спроектировать эндпоинты для протокола A2A

   • POST /register – регистрация агента в реестре (передаётся AgentManifest).
   • GET /discover?capabilities=translate,summarize – поиск агентов по списку требуемых capabilities.
   • POST /negotiate – начало переговоров между агентами (вызывается от agentA к agentB или через реестр).

3. Зафиксировать схему в формате OpenAPI (опционально) или Pydantic models.

Ожидаемый результат этапа Документ с описанием протокола (в README или Markdown) и файл schemas.py с Pydantic моделями.

Этап 2: Реализация реестра агентов (1.5 часа)

Действия

1. Создать FastAPI-приложение с эндпоинтами

   • POST /register – принимает AgentManifest, сохраняет в in-memory dict (или SQLite).
   • GET /discover – принимает query-параметр capabilities (через запятую), фильтрует зарегистрированных агентов и возвращает тех, чьи capabilities содержат все запрошенные.
   • DELETE /agents/{agent_id} – для отмены регистрации.
   • GET /agents – список всех агентов (для отладки).

2. Реализовать проверку уникальности agent_id

3. Добавить время жизни (TTL) для записей агентов – использовать фоновую задачу для удаления устаревших записей (например, каждые 30 секунд).

4. Добавить базовое логирование всех запросов

# Пример модели
from pydantic import BaseModel, Field
from typing import List, Dict, Optional

class AgentManifest(BaseModel):
    agent_id: str
    name: str
    endpoint: str  # URL для прямого вызова агента
    capabilities: List[str]

class DiscoveryResponse(BaseModel):
    agents: List[AgentManifest]

Ожидаемый результат этапа Работающий микросервис реестра с тестовыми запросами через curl / httpx.

Этап 3: Реализация базового агента с A2A-клиентом (2 часа)

Действия

1. Разработать класс Agent с основными методами:

   • register(registry_url) – регистрация в реестре.
   • discover(registry_url, required_capabilities) – поиск других агентов.
   • negotiate(target_agent_url, capabilities, parameters) – отправка negotiation-запроса.
   • handle_negotiation(request) – обработка входящего negotiation-запроса.

2. Реализовать асинхронный HTTP-клиент на aiohttp

3. Реализовать два экземпляра агента с разными capabilities:

   • Агент А – capabilities: ["translate", "summarize"].
   • Агент Б – capabilities: ["translate", "code_review"].
   • Оба запускаются в отдельных процессах/контейнерах с разными портами.

async def negotiate(self, target_url: str, desired_capabilities: List[str]) -> Optional[dict]:
    payload = NegotiationRequest(
        agent_id=self.manifest.agent_id,
        session_id=str(uuid4()),
        capabilities=desired_capabilities,
        parameters={"language": "ru"}
    )
    async with aiohttp.ClientSession() as session:
        async with session.post(f"{target_url}/negotiate", json=payload.dict()) as response:
            if response.status == 200:
                return await response.json()
    return None

Ожидаемый результат этапа Два работающих агента, которые могут зарегистрироваться в реестре и обмениваться HTTP-запросами.

Этап 4: Интеграция discovery и negotiation (1.5 часа)

Действия

1. Написать скрипт, который симулирует use-case:

   • Запустить реестр.
   • Зарегистрировать агента А и агента Б.
   • Агент А хочет найти агента с capability code_review.
   • Агент А запрашивает реестр, получает информацию об агенте Б.
   • Агент А отправляет negotiation-запрос агенту Б.
   • Агент Б обрабатывает запрос, проверяет свои capabilities, соглашается.
   • Вывести логи handshake.

2. Добавить обработку ошибок

   • Если реестр недоступен – повтор с exponential backoff.
   • Если агент не ответил на negotiation (timeout) – переключиться на другого кандидата.
   • Если capabilities не совпадают – возвращать отказ.

3. Записать tracelog всех сообщений (пример ниже).

[REGISTRY] 10:00:00.123 Register agent_A ['translate', 'summarize'] -> OK
[AGENT_A] 10:00:01.456 Discover for ['code_review'] -> found agent_B (http://agent_B:8001)
[AGENT_A] 10:00:02.789 Sent negotiation to agent_B (session_id: abc-123)
[AGENT_B] 10:00:02.790 Received negotiation, checking capabilities: code_review -> accepted
[AGENT_B] 10:00:02.891 Sent negotiation response: accepted
[AGENT_A] 10:00:02.894 Handshake completed with agent_B

Ожидаемый результат этапа Полный цикл discovery + negotiation, зафиксированный в логах.

Этап 5: Покрытие тестами и контейнеризация (1.5 часа)

Действия

1. Написать юнит-тесты для Pydantic моделей и валидации (pytest).

2. Написать интеграционные тесты

   • Тест: регистрация, поиск по существующей capability, поиск по отсутствующей.
   • Тест: negotiation success и failure.
   • Тест: TTL агентов (дождаться удаления и проверить, что discovery возвращает пустой список).

3. Создать docker-compose.yml

   • registry (образ FastAPI, порт 8000).
   • agent_a (тот же образ с агентами или отдельный скрипт, порт 8001).
   • agent_b (порт 8002).
   • Использовать зависимости (depends_on) для порядка запуска.

4. Добавить README с инструкцией запуска

Ожидаемый результат этапа Все тесты проходят, docker-compose up запускает систему из 3 контейнеров.

5. Критерии приемки (Definition of Done)

• Опубликовано описание протокола A2A в формате Markdown (эндпоинты, схемы данных).
• Реестр агентов реализован как отдельный микросервис, принимает регистрацию и возвращает список по capabilities.
• Агент A может зарегистрироваться в реестре.
• Агент B может зарегистрироваться в реестре с другим набором capabilities.
• Агент A успешно находит агента B через discovery-запрос с фильтром ['code_review'].
• Negotation (handshake) между A и B завершается успешно с возвратом согласованных параметров.
• Тайм-ауты и отказ capabilities обрабатываются корректно (лог errors без падений).
• Написаны минимум 3 интеграционных теста (pytest) на основной сценарий.
• Docker-compose поднимает всю систему одной командой.
• В README описаны шаги запуска и пример команд curl.

6. Ожидаемый результат

• В репозитории (папка a2a-protocol):

  • schemas.py – Pydantic модели.
  • registry/ – код реестра (FastAPI).
  • agent.py – класс агента с методами register/discover/negotiate.
  • demo.py – скрипт, запускающий двух агентов и демонстрирующий сценарий.
  • tests/ – тесты.
  • docker-compose.yml – оркестрация.
  • README.md – описание протокола и инструкция.

• Содержимое артефакта

  • Рабочая команда docker-compose up --build запускает реестр и двух агентов.
  • При выполнении demo.py (или аналогичного скрипта) в выводе видны логи discovery и negotiation.
  • Тесты pytest tests/ – зеленые.

• Опционально простой Web UI (на streamlit или fastapi-admin) для просмотра зарегистрированных агентов и их capabilities.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание Для первого выполнения задачи (включая изучение протокола, отладку) закладывайте до 10 часов.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я проверил, что агент A успешно регистрируется в реестре (код 200).
• Я проверил, что при поиске ['code_review'] агент A получает агента B.
• Я проверил, что negotiation-запрос отправляется на правильный endpoint агента B.
• Я проверил, что в логах отображаются все ключевые шаги: регистрация, discovery, handshake.
• Я проверил, что docker-compose up запускает всю систему без ошибок и скрипт demo.py отрабатывает без сбоев.