Перейти к содержанию

Webhooks

Webhooks позволяют внешним системам отправлять события агентов в AegisAI через HTTP POST с криптографической подписью.

Эндпоинт: POST /webhooks/agent

Аутентификация

Каждый webhook-запрос требует два заголовка:

Заголовок Описание
X-API-Key API-ключ арендатора
X-Webhook-Signature HMAC-SHA256 подпись тела запроса
Content-Type application/json

Секрет подписи задаётся переменной WEBHOOK_SECRET.

Формат события

{
  "event": "agent.action",
  "payload": {
    "agent_id": "uuid",
    "action": "webhook_test",
    "resource": "external_system",
    "input": {"source": "my_app"},
    "reasoning": ["Шаг 1", "Шаг 2"],
    "risk_score": 0.02
  }
}

Типы событий

Событие Описание
agent.action Действие агента (логируется в аудит)
agent.heartbeat Сигнал жизни агента

agent.action

{
  "event": "agent.action",
  "payload": {
    "agent_id": "<uuid>",
    "action": "process_order",
    "resource": "orders_db",
    "input": {"order_id": "12345"},
    "output": {"status": "completed"},
    "reasoning": ["Валидация", "Обработка"],
    "risk_score": 0.05
  }
}

agent.heartbeat

{
  "event": "agent.heartbeat",
  "payload": {
    "agent_id": "<uuid>",
    "status": "active"
  }
}

Подпись запроса

Python

import json
import httpx
from aegisai.integrations.webhook.signature import sign_payload

WEBHOOK_SECRET = "webhook-secret-key"
API_KEY = "aegis_your_key"

payload = {
    "event": "agent.action",
    "payload": {
        "agent_id": "agent-uuid",
        "action": "test",
        "resource": "test_resource",
    },
}

body = json.dumps(payload).encode()
signature = sign_payload(body, WEBHOOK_SECRET)

response = httpx.post(
    "http://localhost:8000/webhooks/agent",
    content=body,
    headers={
        "X-API-Key": API_KEY,
        "X-Webhook-Signature": signature,
        "Content-Type": "application/json",
    },
)

Алгоритм подписи

signature = HMAC-SHA256(WEBHOOK_SECRET, request_body)
header_value = "sha256=" + hex(signature)

Безопасность

  • Используйте уникальный WEBHOOK_SECRET для каждого окружения
  • Никогда не передавайте секрет в URL или query-параметрах
  • Проверяйте подпись на стороне получателя

Верификация (на стороне AegisAI)

AegisAI автоматически верифицирует подпись через WebhookReceiver:

  1. Извлекает X-Webhook-Signature из заголовка
  2. Вычисляет HMAC-SHA256 от тела запроса
  3. Сравнивает с переданной подписью
  4. При несовпадении возвращает 401 Invalid webhook signature

Ответы

Успех

{"status": "ok", "event": "agent.action"}

Ошибки

Код Причина
400 Неизвестный тип события
401 Отсутствует или неверная подпись
401 Неверный API-ключ

Полный пример

См. examples/webhook_agent.py:

export AEGISAI_API_KEY=aegis_acme_demo_seed_key_001
export WEBHOOK_SECRET=webhook-secret-key
poetry run python examples/webhook_agent.py

WebhookReceiver (для кастомных интеграций)

from aegisai.integrations.webhook import WebhookReceiver, create_webhook_router

# Создание роутера для FastAPI
router = create_webhook_router(secret_key="your-secret")

# Или прямое использование
receiver = WebhookReceiver(secret_key="your-secret")
is_valid = receiver.verify(body_bytes, signature_header)

Когда использовать Webhooks vs SDK

Критерий SDK Webhooks
Язык Python Любой (HTTP)
Направление Pull (агент → API) Push (внешняя система → API)
Политики Встроенная проверка Только логирование
Сложность Низкая Средняя (подпись)
Use case Python-агенты Микросервисы, serverless, внешние системы