Webhooks¶

AegisAI accepts webhook events from external agents at POST /webhooks/agent. This enables event-driven integration without the Python SDK — ideal for agents written in other languages or running in serverless environments.

Endpoint¶

POST http://localhost:8000/webhooks/agent

Authentication¶

Webhooks use HMAC-SHA256 signature verification:

Header	Value
`X-Webhook-Signature`	HMAC-SHA256 hex digest of the raw request body
`Content-Type`	`application/json`
`X-API-Key`	Tenant API key (for tenant resolution)

The signing secret is configured via WEBHOOK_SECRET (default: webhook-secret-key).

Change the default secret

Set WEBHOOK_SECRET to a strong random value in production.

Payload Format¶

{
  "event": "agent.action",
  "payload": {
    "agent_id": "550e8400-e29b-41d4-a716-446655440000",
    "action": "process_request",
    "resource": "customer_data",
    "input": {"query": "lookup customer"},
    "output": {"found": true},
    "reasoning": ["Received request", "Looked up customer"],
    "risk_score": 0.1,
    "is_approved": true
  }
}

Event Types¶

Event	Description	Handler
`agent.action`	Agent performed an action	Creates audit log entry
`agent.heartbeat`	Agent heartbeat	Updates agent status

agent.action¶

Logs an action to the audit trail. The agent must already be registered.

Payload fields:

Field	Required	Type	Description
`agent_id`	Yes	UUID	Registered agent ID
`action`	No	string	Action name (default: `webhook_action`)
`resource`	No	string	Resource identifier (default: `webhook`)
`input`	No	object	Action input data
`output`	No	object	Action output data
`reasoning`	No	array	Reasoning steps
`risk_score`	No	float	Risk score 0.0–1.0
`is_approved`	No	boolean	Approval status (default: `true`)

agent.heartbeat¶

Updates agent heartbeat status.

Payload fields:

Field	Required	Type	Description
`agent_id`	Yes	UUID	Registered agent ID
`status`	No	string	`active`, `inactive`, or `error`

Signing Payloads¶

Python¶

import hashlib
import hmac
import json
import httpx

WEBHOOK_SECRET = "webhook-secret-key"
BASE_URL = "http://localhost:8000"

def sign_payload(body: bytes, secret: str) -> str:
    return hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()

async def send_webhook(event: str, payload: dict, api_key: str):
    body = json.dumps({"event": event, "payload": payload}).encode()
    signature = sign_payload(body, WEBHOOK_SECRET)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}/webhooks/agent",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Webhook-Signature": signature,
                "X-API-Key": api_key,
            },
        )
        return response.json()

Using the SDK Utility¶

from aegisai.integrations.webhook.signature import sign_payload, verify_signature

body = b'{"event": "agent.action", "payload": {...}}'
signature = sign_payload(body, "webhook-secret-key")

# Verify on receipt
is_valid = verify_signature(body, "webhook-secret-key", signature)

Complete Example¶

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

async def main():
    payload = {
        "event": "agent.action",
        "payload": {
            "agent_id": "your-agent-uuid",
            "action": "webhook_test",
            "resource": "test_resource",
            "input": {"source": "webhook_example"},
            "reasoning": ["Step 1: Send webhook", "Step 2: Verify receipt"],
        },
    }

    body = json.dumps(payload).encode()
    signature = sign_payload(body, "webhook-secret-key")

    async with httpx.AsyncClient() as client:
        response = await client.post(
            "http://localhost:8000/webhooks/agent",
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Webhook-Signature": signature,
                "X-API-Key": "your-tenant-api-key",
            },
        )
        print(response.status_code, response.json())

asyncio.run(main())

See examples/webhook_agent.py in the repository for a runnable example.

Response¶

Success (200):

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

Error Responses¶

Status	Detail	Cause
401	Missing X-Webhook-Signature header	No signature provided
401	Invalid webhook signature	Signature mismatch
400	Unknown event type	Unsupported `event` value
404	Agent not found	Invalid `agent_id`

Security Considerations¶

Best practices

Always sign payloads — never send unsigned webhooks in production
Use HTTPS — protect payloads in transit
Rotate secrets — change WEBHOOK_SECRET periodically
Validate agent_id — register agents before sending webhooks
Idempotency — design handlers to tolerate duplicate deliveries

Webhook vs SDK¶

Feature	Webhook	Python SDK
Language	Any	Python only
Policy check	Manual (call API separately)	Built-in `check_policy()`
Audit logging	Via `agent.action` event	Via `log_action()`
Heartbeat	Via `agent.heartbeat` event	Via `heartbeat()`
Auth	HMAC signature + API key	API key only
Real-time	Fire-and-forget	Async request/response