Agent-to-Agent Protocol (A2A)

{
  "name": "Research Agent",
  "description": "Агент для исследования тем и сбора информации из открытых источников",
  "url": "https://research-agent.example.com",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "authentication": {
    "schemes": ["Bearer"]
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain", "application/json"],
  "skills": [
    {
      "id": "web-research",
      "name": "Web Research",
      "description": "Поиск и анализ информации в интернете",
      "tags": ["research", "web", "analysis"],
      "examples": [
        "Исследуй тему 'квантовые компьютеры в 2025'",
        "Найди статистику по рынку AI в России"
      ]
    },
    {
      "id": "summarize",
      "name": "Document Summarization",
      "description": "Суммаризация документов и статей",
      "tags": ["summary", "documents"]
    }
  ]
}

import httpx
import json

A2A_SERVER = "https://research-agent.example.com"

async def send_task(description: str, session_id: str | None = None):
    """Отправка задачи агенту через A2A."""
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tasks/send",
        "params": {
            "id": "task-001",
            "sessionId": session_id or "session-001",
            "message": {
                "role": "user",
                "parts": [
                    {
                        "type": "text",
                        "text": description
                    }
                ]
            }
        }
    }

    async with httpx.AsyncClient() as client:
        response = await client.post(
            A2A_SERVER,
            json=payload,
            headers={"Authorization": "Bearer <token>"}
        )
        return response.json()

# Использование
result = await send_task(
    "Собери информацию о топ-5 AI-фреймворках для агентов в 2025 году"
)

# Ответ содержит Task с текущим статусом
# {
#   "jsonrpc": "2.0",
#   "id": 1,
#   "result": {
#     "id": "task-001",
#     "sessionId": "session-001",
#     "status": {"state": "working"},
#     "artifacts": []
#   }
# }

import httpx
import json

async def subscribe_to_task(task_id: str):
    """Подписка на обновления задачи через SSE."""
    payload = {
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tasks/sendSubscribe",
        "params": {
            "id": task_id,
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": "Начни исследование"}]
            }
        }
    }

    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            A2A_SERVER,
            json=payload,
            headers={"Accept": "text/event-stream"}
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    event = json.loads(line[6:])
                    result = event.get("result", {})

                    # Обработка обновлений статуса
                    if "status" in result:
                        state = result["status"]["state"]
                        print(f"Статус: {state}")

                        if state == "completed":
                            for artifact in result.get("artifacts", []):
                                for part in artifact.get("parts", []):
                                    print(f"Результат: {part.get('text', '')}")
                            break

                        elif state == "input-required":
                            msg = result["status"].get("message", {})
                            print(f"Агент спрашивает: {msg}")
                            # Здесь нужно отправить ответ
                            break

                        elif state == "failed":
                            msg = result["status"].get("message", {})
                            print(f"Ошибка: {msg}")
                            break

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import json
import asyncio

app = FastAPI()

# Agent Card
AGENT_CARD = {
    "name": "Summary Agent",
    "description": "Суммаризация текстов",
    "url": "https://summary.example.com",
    "version": "1.0.0",
    "capabilities": {"streaming": True},
    "defaultInputModes": ["text/plain"],
    "defaultOutputModes": ["text/plain"],
    "skills": [{
        "id": "summarize",
        "name": "Summarize",
        "description": "Создать краткое содержание текста"
    }]
}

# Хранилище задач (в продакшене — база данных)
tasks: dict = {}

@app.get("/.well-known/agent.json")
async def agent_card():
    return AGENT_CARD

@app.post("/")
async def handle_jsonrpc(request: Request):
    body = await request.json()
    method = body.get("method")

    if method == "tasks/send":
        return await handle_task_send(body)
    elif method == "tasks/get":
        return await handle_task_get(body)
    # ... другие методы

async def handle_task_send(body: dict):
    params = body["params"]
    task_id = params["id"]
    message = params["message"]
    text = message["parts"][0]["text"]

    # Создаём задачу
    tasks[task_id] = {
        "id": task_id,
        "status": {"state": "working"},
        "artifacts": []
    }

    # Выполняем суммаризацию (в реальности — вызов LLM)
    summary = f"Краткое содержание: {text[:100]}..."
    tasks[task_id]["status"]["state"] = "completed"
    tasks[task_id]["artifacts"] = [{
        "parts": [{"type": "text", "text": summary}]
    }]

    return {
        "jsonrpc": "2.0",
        "id": body["id"],
        "result": tasks[task_id]
    }