Внедрение A2A в ваше приложение
Введение
Протокол Agent2Agent (A2A) (Агент-Агент) - это открытый стандарт, инициированный Google, разработанный для обеспечения коммуникации и совместимости между различными системами AI-агентов (искусственного интеллекта). Этот учебник проведет вас через процесс внедрения протокола A2A в ваше приложение, позволяя вашим агентам коммуницировать с другими агентами независимо от используемых фреймворков или вендоров.
Что такое протокол A2A?
Протокол A2A предоставляет стандартизированный способ для AI-агентов (искусственного интеллекта): - Обнаруживать возможности друг друга - Согласовывать режимы взаимодействия (текст, формы, файлы и т.д.) - Безопасно сотрудничать над задачами - Обмениваться информацией в структурированном формате
В своей основе A2A использует JSON-RPC 2.0 через HTTP(S) для коммуникации, поддерживая как стандартные шаблоны запрос/ответ, так и Server-Sent Events (SSE) для потоковой передачи обновлений.
Предварительные требования
Перед внедрением A2A в ваше приложение убедитесь, что у вас есть:
- Базовое понимание HTTP и JSON-RPC
- Знакомство с библиотеками HTTP клиент/сервер вашего языка программирования
- AI-агент (искусственный интеллект) или приложение, которому необходимо взаимодействовать с другими агентами
Шаг 1: Понимание основных концепций
Прежде чем погрузиться в реализацию, давайте поймем ключевые компоненты протокола A2A:
Agent Card (Карточка агента)
Agent Card - это публичный метаданные-файл (обычно расположенный по пути /.well-known/agent.json), который описывает возможности агента, навыки, URL-конечной точки и требования к аутентификации. Он служит механизмом обнаружения для клиентов, чтобы понять, что агент может делать.
A2A Server (Сервер A2A)
A2A Server - это агент, который предоставляет HTTP-конечную точку, реализующую методы протокола A2A. Он принимает запросы и управляет выполнением задач.
A2A Client (Клиент A2A)
A2A Client - это приложение или другой агент, который потребляет сервисы A2A, отправляя запросы на A2A Server.
Task (Задача)
Task - это центральная единица работы в A2A. Клиент инициирует задачу, отправляя сообщение, и задачи проходят через различные состояния (submitted, working, input-required, completed, failed, canceled).
Message (Сообщение)
Messages представляют собой коммуникационные обороты между клиентом (роль: "user") и агентом (роль: "agent"). Сообщения содержат Parts.
Part (Часть)
Parts - это фундаментальные единицы содержимого в Messages или Artifacts. Они могут быть: - TextPart: содержимое простого текста - FilePart: содержимое файла с встроенными байтами или URI - DataPart: структурированные данные JSON (например, формы)
Artifact (Артефакт)
Artifacts представляют собой выходные данные, сгенерированные агентом во время задачи, такие как сгенерированные файлы или структурированные данные. Артефакты также содержат Parts.
Шаг 2: Настройка A2A Server
Давайте реализуем A2A Server на Python с использованием фреймворка Starlette:
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from sse_starlette.sse import EventSourceResponse
from starlette.requests import Request
import uvicorn
import json
from typing import AsyncIterable
# Import A2A types (you would typically use the A2A library)
from common.types import (
AgentCard,
A2ARequest,
GetTaskRequest,
SendTaskRequest,
SendTaskStreamingRequest,
TaskManager
)
class A2AServer:
def __init__(
self,
host="0.0.0.0",
port=5000,
endpoint="/",
agent_card=None,
task_manager=None,
):
self.host = host
self.port = port
self.endpoint = endpoint
self.task_manager = task_manager
self.agent_card = agent_card
self.app = Starlette()
# Set up routes
self.app.add_route(self.endpoint, self._process_request, methods=["POST"])
self.app.add_route(
"/.well-known/agent.json", self._get_agent_card, methods=["GET"]
)
def start(self):
# Validate required components
if self.agent_card is None:
raise ValueError("agent_card is not defined")
if self.task_manager is None:
raise ValueError("task_manager is not defined")
# Start the server
uvicorn.run(self.app, host=self.host, port=self.port)
def _get_agent_card(self, request: Request) -> JSONResponse:
# Return the agent card as JSON
return JSONResponse(self.agent_card.model_dump(exclude_none=True))
async def _process_request(self, request: Request):
try:
# Parse the JSON-RPC request
body = await request.json()
json_rpc_request = A2ARequest.validate_python(body)
# Route to the appropriate handler based on request type
if isinstance(json_rpc_request, GetTaskRequest):
result = await self.task_manager.on_get_task(json_rpc_request)
elif isinstance(json_rpc_request, SendTaskRequest):
result = await self.task_manager.on_send_task(json_rpc_request)
elif isinstance(json_rpc_request, SendTaskStreamingRequest):
result = await self.task_manager.on_send_task_subscribe(
json_rpc_request
)
# Handle other request types...
# Return the result
if isinstance(result, AsyncIterable):
return EventSourceResponse(result)
else:
return JSONResponse(result.model_dump(exclude_none=True))
except Exception as e:
# Handle errors
return JSONResponse({"error": str(e)}, status_code=500)
Шаг 3: Создание менеджера задач
Task Manager (Менеджер задач) обрабатывает фактическую обработку задач. Вот упрощенная реализация:
import asyncio
from typing import AsyncIterable, Dict
from common.types import (
Task,
TaskSendParams,
TaskStatus,
TaskState,
SendTaskResponse,
GetTaskResponse,
SendTaskStreamingResponse
)
class InMemoryTaskManager:
def __init__(self):
self.tasks = {}
self.lock = asyncio.Lock()
async def upsert_task(self, task_send_params: TaskSendParams) -> Task:
async with self.lock:
task = self.tasks.get(task_send_params.id)
if task is None:
# Create a new task
task = Task(
id=task_send_params.id,
sessionId=task_send_params.sessionId,
messages=[task_send_params.message],
status=TaskStatus(state=TaskState.SUBMITTED),
history=[task_send_params.message],
)
self.tasks[task_send_params.id] = task
else:
# Update existing task
task.history.append(task_send_params.message)
return task
async def on_send_task(self, request) -> SendTaskResponse:
# Validate the request
# Process the task
# Return a response
pass
async def on_get_task(self, request) -> GetTaskResponse:
# Retrieve and return task status
pass
async def on_send_task_subscribe(self, request) -> AsyncIterable[SendTaskStreamingResponse]:
# Set up streaming response
# Yield updates as they occur
pass
Шаг 4: Определение логики вашего агента
Теперь давайте создадим пользовательский Task Manager, который реализует логику вашего агента:
class MyAgentTaskManager(InMemoryTaskManager):
def __init__(self, agent):
super().__init__()
self.agent = agent
async def on_send_task(self, request):
# Validate the request
error = self._validate_request(request)
if error:
return error
# Create or update the task
await self.upsert_task(request.params)
# Process the task with your agent
task_id = request.params.id
user_message = request.params.message
# Extract the user query
user_query = self._get_user_query(request.params)
# Process with your agent
response = self.agent.process(user_query)
# Update task status and create response
task = await self._update_task_with_response(task_id, response)
# Return the response
return SendTaskResponse(
id=request.id,
result=task
)
Шаг 5: Настройка A2A Client
Для коммуникации с A2A серверами вам понадобится A2A клиент:
import httpx
from typing import Any, AsyncIterable
from common.types import (
AgentCard,
GetTaskRequest,
SendTaskRequest,
SendTaskResponse,
JSONRPCRequest
)
import json
class A2AClient:
def __init__(self, agent_card=None, url=None):
if agent_card:
self.url = agent_card.url
elif url:
self.url = url
else:
raise ValueError("Must provide either agent_card or url")
async def send_task(self, payload: dict[str, Any]) -> SendTaskResponse:
request = SendTaskRequest(params=payload)
return SendTaskResponse(**await self._send_request(request))
async def _send_request(self, request: JSONRPCRequest) -> dict[str, Any]:
async with httpx.AsyncClient() as client:
try:
response = await client.post(
self.url, json=request.model_dump(), timeout=30
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
raise Exception(f"HTTP error: {e}")
except json.JSONDecodeError as e:
raise Exception(f"JSON decode error: {e}")
Шаг 6: Обнаружение агентов с помощью карточек агентов
Для обнаружения возможностей агента вы можете получить его Agent Card (Карточку агента):
class A2ACardResolver:
def __init__(self, base_url, agent_card_path="/.well-known/agent.json"):
self.base_url = base_url.rstrip("/")
self.agent_card_path = agent_card_path.lstrip("/")
def get_agent_card(self) -> AgentCard:
with httpx.Client() as client:
response = client.get(f"{self.base_url}/{self.agent_card_path}")
response.raise_for_status()
try:
return AgentCard(**response.json())
except json.JSONDecodeError as e:
raise Exception(f"JSON decode error: {e}")
Шаг 7: Собираем все вместе
Теперь давайте создадим полный пример, который настраивает A2A сервер с простым агентом:
from common.types import AgentCard, Capability, Skill
# Define your agent
class MyAgent:
SUPPORTED_CONTENT_TYPES = ["text"]
def process(self, query):
# Your agent logic here
return f"Processed: {query}"
# Define capabilities and skills
capabilities = [
Capability(
type="text-generation",
description="Generates text responses to user queries"
)
]
skills = [
Skill(
name="text-response",
description="Responds to text queries with generated text"
)
]
# Create the agent card
agent_card = AgentCard(
name="My A2A Agent",
description="A simple A2A-compatible agent",
url="http://localhost:5000/",
version="1.0.0",
defaultInputModes=MyAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=MyAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=skills,
)
# Create and start the server
server = A2AServer(
agent_card=agent_card,
task_manager=MyAgentTaskManager(agent=MyAgent()),
host="localhost",
port=5000,
)
server.start()
Шаг 8: Использование вашего A2A агента
Для использования вашего A2A агента из другого приложения:
import asyncio
from uuid import uuid4
async def main():
# Create a client
client = A2AClient(url="http://localhost:5000/")
# Generate a unique task ID
task_id = str(uuid4())
# Create a task payload
payload = {
"id": task_id,
"sessionId": str(uuid4()),
"acceptedOutputModes": ["text"],
"message": {
"role": "user",
"parts": [
{
"type": "text",
"text": "Hello, agent!",
}
],
},
}
# Send the task
response = await client.send_task(payload)
# Print the response
print(f"Response: {response}")
# Run the client
asyncio.run(main())
Расширенные функции
Потоковые ответы
Для долговыполнющихся задач, вы можете использовать потоковую передачу для получения обновлений в реальном времени:
async def stream_task():
client = A2AClient(url="http://localhost:5000/")
task_id = str(uuid4())
stream_params = {
"id": task_id,
"message": {
"role": "user",
"parts": [{"text": "Stream me some updates!"}]
},
}
stream = client.send_task_subscribe(stream_params)
async for event in stream:
# Process streaming events
print(f"Received event: {event}")
Обработка файлов
Для обмена файлами между агентами:
# Sending a file
file_payload = {
"id": task_id,
"message": {
"role": "user",
"parts": [
{
"type": "file",
"file": {
"mime_type": "image/jpeg",
"file_name": "image.jpg",
"data": base64.b64encode(file_data).decode("utf-8")
}
}
]
}
}
# Receiving a file
response = await client.send_task(file_payload)
for artifact in response.result.artifacts:
for part in artifact.parts:
if part.type == "file":
file_data = base64.b64decode(part.file.data)
# Process the file
Заключение
Протокол A2A предоставляет стандартизированный способ для AI-агентов (искусственного интеллекта) коммуницировать и сотрудничать, независимо от их базовых фреймворков или вендоров. Внедряя A2A в ваше приложение, вы позволяете вашим агентам участвовать в более широкой экосистеме совместимых AI-систем (искусственного интеллекта).
Этот учебник охватил основы внедрения A2A в вашем приложении, включая настройку A2A сервера, создание менеджера задач, определение логики агента и использование A2A клиента. Для более продвинутых функций и подробной документации обратитесь к официальной документации протокола A2A.
Ресурсы
- A2A Protocol GitHub Repository
- A2A Protocol Documentation
- A2A Protocol Specification