Проблема: агенты без контракта
Представь команду из пяти агентов: один занимается анализом данных, другой — написанием кода, третий — поиском в интернете, четвёртый — работой с документами, пятый — оркестрирует остальных. Как оркестратор знает, кому передать задачу «построй график по CSV»?
Без формальной спецификации — никак. Он либо знает это из хардкода («если в задаче слово "данные" — вызывай агент А»), либо спрашивает LLM угадать. Оба подхода ломаются при масштабировании: добавился новый агент — нужно переписывать оркестратор. Изменились возможности существующего агента — оркестратор об этом не знает.
Agent Card решает задачу через машиночитаемое описание: каждый агент публикует JSON-документ с тем, что он умеет. Оркестратор читает карточки всех доступных агентов и принимает решение о маршрутизации на основе данных, а не хардкода.
БЕЗ Agent Card: С Agent Card:
Оркестратор Оркестратор
│ │
├── if "данные" → агент А │ читает /.well-known/agent.json
├── if "код" → агент Б │ каждого агента в сети
├── if "поиск" → агент В │
└── (hardcoded, хрупко) ├── агент А: skills=[data-analysis, charting]
├── агент Б: skills=[code-generation, debugging]
Добавить агента Г? ├── агент В: skills=[web-search, summarization]
→ переписывать оркестратор │
└── динамически выбирает по skills
Анатомия Agent Card: из чего состоит документ
Agent Card — это JSON-файл, который обычно хостится по адресу
/.well-known/agent.json на сервере агента.
Он состоит из четырёх логических блоков:
→ кто этот агент и где его найти
→ что конкретно умеет делать (список именованных возможностей)
→ технические возможности: поддерживает ли стриминг, колбэки, историю
→ как аутентифицироваться: Bearer, API Key, OAuth2, none
→ форматы по умолчанию: text, file, data (JSON), audio, image
Полный Agent Card в JSON — вот как это выглядит для агента-аналитика данных:
{
"name": "Data Analysis Agent",
"description": "Анализирует структурированные данные, строит графики и пишет отчёты. Работает с CSV, JSON и SQL-источниками.",
"url": "https://agents.example.com/data-analyst",
"version": "1.2.0",
"provider": {
"organization": "Example Corp",
"url": "https://example.com"
},
"defaultInputModes": ["text", "data", "file"],
"defaultOutputModes": ["text", "data", "file"],
"capabilities": {
"streaming": true,
"pushNotifications": false,
"stateTransitionHistory": true
},
"authentication": {
"schemes": ["Bearer"],
"credentials": null
},
"skills": [
{
"id": "data-analysis",
"name": "Анализ данных",
"description": "Статистический анализ датасетов: дескриптивная статистика, выбросы, корреляции, тренды. Принимает CSV/JSON, возвращает структурированный отчёт.",
"tags": ["analytics", "statistics", "csv", "json"],
"examples": [
"Проанализируй продажи за Q3 и найди аномалии",
"Посчитай корреляцию между метриками retention и revenue"
],
"inputModes": ["text", "file", "data"],
"outputModes": ["text", "data"]
},
{
"id": "charting",
"name": "Построение графиков",
"description": "Строит визуализации: линейные, столбчатые, тепловые карты, scatter plots. Возвращает PNG или интерактивный HTML.",
"tags": ["visualization", "charts", "matplotlib", "plotly"],
"examples": [
"Построй график продаж по месяцам с трендовой линией",
"Сделай heatmap корреляций для таблицы metrics"
],
"inputModes": ["text", "data"],
"outputModes": ["file", "text"]
},
{
"id": "sql-query",
"name": "SQL-запросы",
"description": "Пишет и выполняет SELECT-запросы к подключённым базам данных. Только чтение, DDL запрещён.",
"tags": ["sql", "database", "postgresql", "sqlite"],
"examples": [
"Найди топ-10 пользователей по выручке за последний месяц",
"Сколько заказов в статусе pending больше 3 дней"
],
"inputModes": ["text"],
"outputModes": ["text", "data"]
}
]
}
Skills: как описывать что умеет агент
Skill — это именованная возможность агента, которую можно вызвать как единицу.
Она отличается от инструмента (tool): инструмент — это конкретная функция
(например, read_file), skill — это высокоуровневая задача
(например, «анализ данных»), которая может использовать множество инструментов внутри.
Хорошая skill содержит пять элементов:
| Поле | Назначение и правила | |
|---|---|---|
| id | req | Уникальный идентификатор в kebab-case. Используется при маршрутизации: оркестратор сопоставляет задачу с id. Не меняй между версиями — это breaking change. |
| name | req | Человекочитаемое название. Отображается в UI и логах. |
| description | req | Описание для LLM-роутера. Пиши что делает, с чем работает, что не умеет. Это основной текст, по которому LLM решает — подходит ли агент для задачи. |
| examples | opt | Примеры запросов (1–3 фразы). Критически важны для embedding-based роутинга: чем лучше примеры, тем точнее поиск похожих задач. |
| tags | opt | Ключевые слова для фильтрации. Позволяют грубо отсеять неподходящих агентов перед точным LLM-матчингом. |
| inputModes / outputModes | opt | Форматы для этого конкретного skill. Переопределяют defaultInputModes на уровне карточки. |
Примеры хорошо и плохо описанных skills:
Capabilities: технические возможности
Если skills описывают что умеет агент, capabilities описывают как он это делает с технической точки зрения. Эти флаги нужны клиенту, чтобы правильно настроить взаимодействие.
true — клиент может использовать SSE или WebSocket вместо ожидания полного ответа.Input и Output modes
Режимы ввода/вывода определяют, в каком формате агент принимает задачи и возвращает результаты. Протокол A2A определяет пять стандартных режимов:
| Режим | Описание |
|---|---|
| text | Обычный текст. Самый распространённый режим — подходит для большинства задач. |
| data | Структурированные данные в виде JSON-объекта. Для передачи таблиц, метрик, конфигураций. |
| file | Бинарные или текстовые файлы: CSV, PNG, PDF, исходный код. Передаются как base64 или URL. |
| audio | Аудиофайлы или аудиопоток. Для voice-агентов. |
| image | Изображения: PNG, JPEG, SVG. Для агентов с vision capabilities. |
text по умолчанию, но skill charting
дополнительно принимает data (датасет для построения графика).
Ограничения: что агент не может
Явные ограничения в Agent Card так же важны, как и skills. Без них оркестратор может отправить задачу «не тому» агенту — тот примет её, начнёт выполнять и провалится в середине работы. Провал в середине дороже, чем отказ на этапе маршрутизации.
Ограничения размещаются в поле description карточки
и в description конкретных skills. Нет стандартного
поля constraints — их вписывают явным текстом.
Вот что стоит документировать:
{
"name": "Code Review Agent",
"description": "Проверяет Python-код на качество, безопасность и стиль. НЕ пишет новый код — только ревьюирует существующий. НЕ работает с TypeScript, Java, Go. Максимальный размер файла: 500 КБ. Не имеет доступа к интернету.",
"skills": [
{
"id": "security-audit",
"name": "Аудит безопасности",
"description": "Ищет уязвимости в Python-коде: инъекции, небезопасные десериализации, hardcoded secrets, небезопасные HTTP-запросы. Анализирует только статически — не запускает код. Возвращает список найденных проблем с severity и рекомендациями по исправлению.",
"tags": ["security", "python", "static-analysis", "owasp"],
"examples": [
"Проверь этот файл на SQL-инъекции",
"Есть ли в коде hardcoded API keys?"
]
}
]
}
Типичные категории ограничений для документирования:
| Категория | Пример в description |
|---|---|
| Языки / форматы | «Работает только с Python и Go. Не поддерживает TypeScript.» |
| Размер данных | «Максимум 100 строк кода или 50 KB текста за один запрос.» |
| Доступ к сети | «Не имеет доступа к интернету. Работает только с переданными данными.» |
| Действия запрещены | «Только чтение — не изменяет файлы и не делает записи в БД.» |
| Внешние зависимости | «Требует подключённую PostgreSQL с таблицей orders.» |
| Latency / SLA | «Среднее время ответа 30–120 сек для больших файлов.» |
Реализация на Python: dataclasses и JSON
Agent Card — просто JSON. Для строгой типизации, валидации и сериализации
используем dataclasses с dataclasses-json или Pydantic.
Здесь — чистый вариант на dataclasses, без лишних зависимостей.
"""
Структуры данных для Agent Card.
Совместимы со спецификацией A2A (Google, 2025).
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field, asdict
from typing import Literal
InputMode = Literal["text", "data", "file", "audio", "image"]
OutputMode = Literal["text", "data", "file", "audio", "image"]
AuthScheme = Literal["Bearer", "ApiKey", "OAuth2", "none"]
@dataclass
class AgentSkill:
"""Одна именованная возможность агента."""
id: str # kebab-case, стабильный идентификатор
name: str # Человекочитаемое название
description: str # Описание для LLM-роутера
tags: list[str] = field(default_factory=list)
examples: list[str] = field(default_factory=list)
input_modes: list[InputMode] = field(default_factory=list)
output_modes: list[OutputMode] = field(default_factory=list)
def __post_init__(self):
if not self.id or not self.id.replace("-", "").isalnum():
raise ValueError(f"Skill id должен быть kebab-case: '{self.id}'")
if not self.description:
raise ValueError(f"Skill '{self.id}' должен иметь description")
@dataclass
class AgentCapabilities:
"""Технические возможности агента."""
streaming: bool = False
push_notifications: bool = False
state_transition_history: bool = False
@dataclass
class AgentAuthentication:
schemes: list[AuthScheme] = field(default_factory=lambda: ["none"])
credentials: str | None = None
@dataclass
class AgentProvider:
organization: str
url: str = ""
@dataclass
class AgentCard:
"""
Полная спецификация агента.
Публикуется по /.well-known/agent.json
"""
name: str
description: str
url: str
skills: list[AgentSkill]
version: str = "1.0.0"
provider: AgentProvider | None = None
default_input_modes: list[InputMode] = field(
default_factory=lambda: ["text"]
)
default_output_modes: list[OutputMode] = field(
default_factory=lambda: ["text"]
)
capabilities: AgentCapabilities = field(
default_factory=AgentCapabilities
)
authentication: AgentAuthentication = field(
default_factory=AgentAuthentication
)
def __post_init__(self):
if not self.skills:
raise ValueError("Agent Card должна содержать хотя бы один skill")
ids = [s.id for s in self.skills]
if len(ids) != len(set(ids)):
raise ValueError("Skill ids должны быть уникальными")
def to_json(self, indent: int = 2) -> str:
"""Сериализует в JSON по спецификации A2A (camelCase)."""
d = asdict(self)
return json.dumps(_to_camel(d), ensure_ascii=False, indent=indent)
@classmethod
def from_json(cls, text: str) -> "AgentCard":
"""Десериализует из JSON."""
d = _from_camel(json.loads(text))
skills = [AgentSkill(**s) for s in d.pop("skills")]
caps = AgentCapabilities(**d.pop("capabilities", {}))
auth = AgentAuthentication(**d.pop("authentication", {}))
prov = AgentProvider(**d.pop("provider")) if d.get("provider") else None
return cls(skills=skills, capabilities=caps, authentication=auth,
provider=prov, **d)
def matches_task(self, task: str) -> list[AgentSkill]:
"""
Простой поиск подходящих skills по ключевым словам задачи.
В реальных системах используется embedding-similarity.
"""
task_lower = task.lower()
matched = []
for skill in self.skills:
score = 0
if any(tag in task_lower for tag in skill.tags):
score += 2
if any(word in task_lower
for word in skill.description.lower().split()
if len(word) > 4):
score += 1
if any(ex_word in task_lower
for ex in skill.examples
for ex_word in ex.lower().split()
if len(ex_word) > 4):
score += 1
if score > 0:
matched.append((score, skill))
return [s for _, s in sorted(matched, reverse=True)]
# ── Утилиты конвертации snake_case ↔ camelCase ───────────────────────
def _to_camel(d):
if isinstance(d, dict):
return {_snake_to_camel(k): _to_camel(v) for k, v in d.items() if v is not None}
if isinstance(d, list):
return [_to_camel(i) for i in d]
return d
def _from_camel(d):
if isinstance(d, dict):
return {_camel_to_snake(k): _from_camel(v) for k, v in d.items()}
if isinstance(d, list):
return [_from_camel(i) for i in d]
return d
def _snake_to_camel(s: str) -> str:
parts = s.split("_")
return parts[0] + "".join(p.capitalize() for p in parts[1:])
def _camel_to_snake(s: str) -> str:
import re
return re.sub(r"(?
Строим карточку и публикуем её через FastAPI:
"""
Агент-аналитик с публикацией Agent Card.
pip install fastapi uvicorn anthropic
"""
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from agent_card import AgentCard, AgentSkill, AgentCapabilities, AgentProvider
# ── Определяем карточку ──────────────────────────────────────────────
CARD = AgentCard(
name="Data Analysis Agent",
description=(
"Анализирует структурированные данные, строит графики и пишет SQL. "
"НЕ пишет код — только анализирует данные. "
"НЕ имеет доступа к интернету. "
"Максимальный размер файла: 10 MB."
),
url="https://agents.example.com/data-analyst",
version="1.2.0",
provider=AgentProvider(organization="Example Corp"),
default_input_modes=["text", "data", "file"],
default_output_modes=["text", "data"],
capabilities=AgentCapabilities(
streaming=True,
state_transition_history=True,
),
skills=[
AgentSkill(
id="data-analysis",
name="Анализ данных",
description=(
"Статистический анализ датасетов: дескриптивная статистика, "
"выбросы, корреляции, тренды. Принимает CSV/JSON. "
"НЕ делает предсказания (для ML используй другой агент)."
),
tags=["analytics", "statistics", "csv", "json", "outliers"],
examples=[
"Проанализируй продажи за Q3 и найди аномалии",
"Посчитай корреляцию между retention и revenue",
],
input_modes=["text", "file", "data"],
output_modes=["text", "data"],
),
AgentSkill(
id="charting",
name="Построение графиков",
description=(
"Строит визуализации: линейные, bar, heatmap, scatter. "
"Возвращает PNG или интерактивный HTML. "
"Требует числовые данные — не работает с текстовыми категориями без маппинга."
),
tags=["visualization", "charts", "matplotlib", "plotly"],
examples=[
"Построй график продаж по месяцам с трендом",
"Heatmap корреляций для таблицы metrics",
],
input_modes=["text", "data"],
output_modes=["file"],
),
],
)
# ── FastAPI ──────────────────────────────────────────────────────────
app = FastAPI(title=CARD.name)
@app.get("/.well-known/agent.json")
async def agent_card():
"""Публикует Agent Card по стандартному пути."""
return JSONResponse(content=CARD.to_json())
@app.post("/tasks/send")
async def handle_task(request: dict):
"""Основной эндпоинт для задач (A2A-совместимый)."""
message = request.get("message", {})
text = ""
for part in message.get("parts", []):
if part.get("type") == "text":
text += part.get("text", "")
# Проверяем, есть ли подходящий skill
matched = CARD.matches_task(text)
if not matched:
return {"status": "rejected",
"reason": "Задача не соответствует ни одному skill этого агента"}
skill = matched[0]
# Здесь запускаем реальную логику агента...
return {
"status": "working",
"matched_skill": skill.id,
"task_id": "task-123"
}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8001)
Маршрутизация на основе Agent Cards
Главная практическая ценность Agent Card — возможность строить динамический роутинг: оркестратор читает карточки всех агентов и выбирает подходящего без хардкода.
GET /.well-known/agent.json у каждого.
Кэширует карточки с TTL (например, 5 минут).
tags в их skills. Отсеивает заведомо неподходящих
без дорогого LLM-вызова.
description + examples,
либо используется cosine similarity между embedding'ами задачи и описаний skills.
streaming: true. Если нужен файловый вывод —
outputModes должен содержать file.
url из карточки
с указанием skill_id. Агент принимает задачу,
проверяет skill и возвращает статус.
"""
Простой маршрутизатор на основе Agent Cards.
pip install httpx anthropic
"""
import httpx
import json
from agent_card import AgentCard
import anthropic
class AgentRouter:
def __init__(self, agent_urls: list[str]):
self.agent_urls = agent_urls
self._cards: dict[str, AgentCard] = {}
self._client = anthropic.Anthropic()
async def discover(self):
"""Загружает Agent Cards со всех агентов."""
async with httpx.AsyncClient(timeout=5.0) as http:
for url in self.agent_urls:
try:
resp = await http.get(f"{url}/.well-known/agent.json")
resp.raise_for_status()
card = AgentCard.from_json(resp.text)
self._cards[url] = card
print(f"Discovered: {card.name} ({len(card.skills)} skills)")
except Exception as e:
print(f"Failed to discover {url}: {e}")
def route(self, task: str) -> tuple[AgentCard, str] | None:
"""
Возвращает (AgentCard, skill_id) для задачи.
Использует LLM для финального выбора.
"""
if not self._cards:
return None
# Строим список кандидатов для LLM
candidates = []
for url, card in self._cards.items():
for skill in card.skills:
candidates.append({
"agent": card.name,
"url": url,
"skill_id": skill.id,
"skill_name": skill.name,
"description": skill.description,
"examples": skill.examples[:2],
})
# Просим LLM выбрать лучший вариант
prompt = f"""Задача пользователя: "{task}"
Доступные агенты и их навыки:
{json.dumps(candidates, ensure_ascii=False, indent=2)}
Выбери наиболее подходящий skill для этой задачи.
Ответь JSON: {{"url": "...", "skill_id": "...", "reason": "..."}}
Если ни один не подходит — {{"url": null, "skill_id": null, "reason": "..."}}"""
resp = self._client.messages.create(
model="claude-opus-4-6",
max_tokens=256,
messages=[{"role": "user", "content": prompt}],
)
try:
choice = json.loads(resp.content[0].text)
except json.JSONDecodeError:
return None
if not choice.get("url"):
print(f"Нет подходящего агента: {choice.get('reason')}")
return None
card = self._cards.get(choice["url"])
if card:
print(f"Роутер выбрал: {card.name} → {choice['skill_id']}")
print(f"Причина: {choice.get('reason')}")
return card, choice["skill_id"]
return None
async def send_task(self, task: str, data: dict | None = None) -> dict:
"""Маршрутизирует задачу и отправляет её подходящему агенту."""
result = self.route(task)
if not result:
return {"error": "Нет подходящего агента для задачи"}
card, skill_id = result
payload = {
"skill_id": skill_id,
"message": {
"role": "user",
"parts": [{"type": "text", "text": task}]
+ ([{"type": "data", "data": data}] if data else [])
}
}
async with httpx.AsyncClient(timeout=30.0) as http:
resp = await http.post(f"{card.url}/tasks/send", json=payload)
return resp.json()
# Использование
import asyncio
async def main():
router = AgentRouter([
"http://localhost:8001", # data-analysis agent
"http://localhost:8002", # code-review agent
"http://localhost:8003", # web-search agent
])
await router.discover()
result = await router.send_task(
"Посчитай корреляцию между метриками в этом датасете",
data={"rows": [{"a": 1, "b": 2}, {"a": 3, "b": 5}]}
)
print(result)
asyncio.run(main())
Типичные ошибки
data-analysis переименовали в analytics в версии 2.0.
Все оркестраторы, которые знали старый id, теперь получают ошибку при попытке
вызвать его явно. Это breaking change.
Шпаргалка
- Что это: JSON-документ по адресу
/.well-known/agent.json— машиночитаемый контракт агента - 4 блока: идентификация (name, url, version) + skills + capabilities + authentication
- Skill: id (kebab-case, стабильный) + name + description (для LLM) + examples + tags + inputModes/outputModes
- Хорошая description skill: что делает + с чем работает + что НЕ умеет
- Capabilities: streaming, pushNotifications, stateTransitionHistory — технические флаги
- Input/output modes: text, data (JSON), file, audio, image
- Ограничения: пиши явно в description — размеры, языки, доступ, запрещённые операции
- Роутинг: 1) обнаружение → 2) фильтр по тегам → 3) LLM/embedding матчинг → 4) проверка capabilities → 5) отправка
- Skill id — это API: не меняй между minor версиями, изменение = breaking change
- Кэш с TTL: карточки кэшировать, но не вечно — агент может обновиться
Практика
AgentCard из урока.
AgentCard метод validate_task(task: str) → ValidationResult,
который возвращает: подходящий skill (или None), причину отказа, список недостающих
inputModes (если задача подразумевает файл, а агент не принимает file).
Напиши 5 тест-кейсов с разными задачами и проверь корректность.
description + все examples.
При поступлении задачи создавай её embedding и ищи ближайший skill
по cosine similarity (через numpy).
Сравни точность (доля правильных роутингов на 20 тест-задачах)
и скорость с LLM-роутером.