aeonic-agentguard-sdk-python

Aeonic Python SDK

Aeonic Python SDK enables runtime monitoring, drift detection, and governance for AI agents running inside Python applications.

It works by observing agent-powered routes at runtime, capturing request/response samples, and securely sending them to the Aeonic platform for analysis.

The SDK is:

• Framework-aware
• Declarative
• Non-intrusive
• Production-safe

---

✨ Supported Frameworks

• FastAPI
• Django

Only routes explicitly marked as agent routes are tracked.
All other routes are completely ignored by the SDK.

---

📦 Installation

Install the base SDK (core only — requests is included):

pip install aeonic-agentguard-sdk-python

Install with your framework (recommended):

# FastAPI
pip install aeonic-agentguard-sdk-python[fastapi]

# Django
pip install aeonic-agentguard-sdk-python[django]

Note: The PyPI package name is aeonic-agentguard-sdk-python, but you import it as aeonic.

Check the installed version:

import aeonic
print(aeonic.__version__)

---

📋 Requirements

Environment variables (optional)

---

🚀 Quick Start (FastAPI)

from fastapi import Depends, FastAPI
from aeonic import init
from aeonic.adapters.fastapi import agent_middleware, with_agent

app = FastAPI()
app.add_middleware(agent_middleware)

init({
    "api_key": "your_api_key",
    "app": app,
    "service_name": "my-service",
})

@app.post(
    "/agent/chat",
    dependencies=[Depends(with_agent({"name": "ChatAgent", "type": "support", "model": ["gpt-4o"]}))],
)
async def chat_agent(payload: dict):
    return {"reply": "Hello!"}

Run with:

uvicorn main:app --reload

---

🔑 SDK Initialization (Required)

Initialize the SDK once at application startup.

Configuration Options

init({
    "api_key": str,                    # REQUIRED
    "app": Any | None,                 # Optional: FastAPI/Django app for route introspection
    "service_name": str | None,        # Optional (recommended)
    "introspection_delay_ms": int,     # Optional: Delay before route scan (default: 2000ms)
    "capture_payloads": {
        "max_samples": int             # Default: 10
    }
})

Example (Basic)

from aeonic import init

init({
    "api_key": "ag_test_123",
    "service_name": "payments-api",
    "capture_payloads": {
        "max_samples": 10
    }
})

Example (With Route Introspection) ⭐ NEW

from aeonic import init
from fastapi import FastAPI

app = FastAPI()

# Pass app instance for automatic route discovery
init({
    "api_key": "ag_test_123",
    "app": app,  # ← Enables automatic agent discovery at startup
    "service_name": "payments-api",
    "introspection_delay_ms": 2000,  # Wait 2s for routes to be defined
    "capture_payloads": {
        "max_samples": 10
    }
})

What this does

• Authenticates your service with Aeonic
• Automatically resolves tenant identity via api_key
• Sets payload sampling limits per agent route
• NEW: Discovers and registers all agent routes at startup (if app is provided)
• NEW: Emits complete agent inventory to the Aeonic platform before serving traffic

📖 Route introspection: Pass your app instance to init() so the SDK can discover agent routes automatically at startup.

---

🚀 FastAPI Integration

Aeonic uses two components in FastAPI:

1. A global middleware (registered once)
2. A route-level dependency to mark agent routes

---

1️⃣ Register Global Middleware

from fastapi import FastAPI
from aeonic.adapters.fastapi import agent_middleware

app = FastAPI()

# 🔴 MUST be registered once
app.add_middleware(agent_middleware)

This middleware:

• Hooks into the response lifecycle
• Does nothing unless the route is marked as an agent route

---

2️⃣ Mark Agent Routes (Required)

Only routes using with_agent() are tracked.

from fastapi import Depends
from aeonic.adapters.fastapi import with_agent

@app.post(
    "/agent/finance",
    dependencies=[
        Depends(
            with_agent({
                "name": "RefundRiskAgent",
                "type": "finance",
                "model": ["gpt-4o"]
            })
        )
    ]
)
async def finance_agent(payload: dict):
    return {
        "approved": True,
        "amount": payload["amount"]
    }

What happens internally

• Request + response payloads are captured
• Samples are buffered per agent route
• Once max_samples is reached, data is sent to the Aeonic platform
• Agent health status is evaluated (healthy / warning / drift)

---

✅ FastAPI Summary

---

🏛️ Django Integration

Django uses:

• A global middleware
• A route decorator to mark agent routes

---

1️⃣ Register Middleware

settings.py

MIDDLEWARE = [
    ...
    "aeonic.adapters.django.agent_middleware",
]

Ensure init() is called during startup (e.g., in settings.py or wsgi.py).

---

2️⃣ Mark Agent Routes with Decorator

from django.http import JsonResponse
from aeonic.adapters.django import with_agent

@with_agent({
    "name": "RefundRiskAgent",
    "type": "finance",
    "model": ["gpt-4o"]
})
def finance_agent(request):
    return JsonResponse({
        "approved": True,
        "amount": 500
    })

---

✅ Django Summary

---

🧠 Agent Concepts

Agent Identity

Each agent route is identified by:

• name – logical agent name
• type – business domain (finance, healthcare, etc.)
• model – AI models used

Route → Agent Mapping

Aeonic correlates:

HTTP Route + Method + Agent Metadata

This enables:

• Drift detection
• Agent behavior consistency checks
• Health classification per route

---

🔐 Security & Privacy

• Payloads are sampled, not streamed
• Only up to max_samples are collected per flush
• SDK never blocks or alters application behavior
• Non-agent routes are never inspected

---

📊 Agent Status Lifecycle

The Aeonic platform classifies agents as:

• Healthy
• Warning
• Drifting
• Failed

Status updates are visible in the Aeonic dashboard.

---

⚠️ Important Rules

• ❌ Do NOT wrap business logic in SDK functions
• ❌ Do NOT rely on thread-local or contextvar hacks
• ✅ Always mark agent routes explicitly
• ✅ Initialize SDK before app starts

---

🔧 Troubleshooting

RuntimeError: Aeonic SDK not initialized

Call init({...}) once at application startup before any SDK features are used.

Samples are not being captured

1. Ensure the route is marked with with_agent() (FastAPI dependency or Django decorator).
2. Ensure global middleware is registered (agent_middleware).
3. Confirm the agent is not blocked or quarantined (check Aeonic dashboard).
4. Set AEONIC_DEBUG=1 to see SDK debug output.

ModuleNotFoundError: No module named 'fastapi' or 'django'

Install the matching framework extra:

pip install aeonic-agentguard-sdk-python[fastapi]
# or
pip install aeonic-agentguard-sdk-python[django]

Agent inventory not emitted at startup

• For FastAPI, pass "app": app to init().
• For Django, ensure init() runs during startup and routes are registered before introspection completes.
• Increase introspection_delay_ms if routes are registered late (e.g. 3000 or 5000).

Platform connection issues

Set your Aeonic endpoint URL (provided during onboarding):

export AEONIC_ENDPOINT=https://your-aeonic-endpoint

On Windows PowerShell:

$env:AEONIC_ENDPOINT="https://your-aeonic-endpoint"

---

📄 License

MIT © Aeonic

---

📚 SDK Function Reference

This section documents the public SDK functions and middleware for integrating Aeonic into your application.

---

aeonic.init(config: dict)

Purpose

Initializes the Aeonic SDK once at application startup, validates and normalizes configuration, kicks off background route introspection, and (optionally) registers the process as an AgentGuard instance.

Arguments

• config (dict) – configuration object with the following keys:
  • api_key (str, required): Your Aeonic API key. If missing, init raises ValueError.
  • app (FastAPI | Django | None, optional): Application instance used for automatic route introspection and agent discovery at startup. Required for FastAPI introspection; optional for Django (Django can be introspected without app).
  • service_name (str | None, optional): Logical name for the service (e.g. "payments-api"). Propagated to all emitted events and inventories.
  • capture_payloads (dict, optional): Controls how request/response payload samples are collected.
    • max_samples (int, optional): Number of samples to buffer per agent route before flushing. Defaults to 10 when omitted.
  • introspection_delay_ms (int, optional): Delay (in milliseconds) before route introspection runs. Defaults to 2000 ms, giving the application time to register all routes.
  • agentguard_enabled (bool, optional): When True (default), the SDK registers itself as an AgentGuard instance in a background thread. When False, this registration step is skipped, but sample collection still works.

Behavior

• Stores a normalized configuration internally containing:
  • api_key, service_name, max_samples, app, introspection_delay_ms, agentguard_enabled.
• Logs a startup message ("[Aeonic] SDK initialized.") using the aeonic logger.
• If agentguard_enabled is True, registers your service with Aeonic in a background daemon thread:
  • Failures are logged as warnings but never raised; the SDK continues collecting samples even if registration fails.
• Schedules deferred route introspection in another background daemon thread:
  • Waits introspection_delay_ms before running.
  • For FastAPI: requires config["app"] and calls introspect_routes(app).
  • For Django: if app is None but Django is installed, calls introspect_routes(None), which uses Django’s URL resolver.
  • On success, emits an agent inventory event with all registered agents (name, type, model, route, method, framework, timestamps).
  • On any error, logs the traceback and falls back to emitting inventory with whatever agents were registered manually.
• If introspection is skipped (no app and Django not installed), calls a fallback that attempts to emit inventory once agents exist.
• Never blocks the main thread; all network calls and introspection work are done in background daemon threads.

Typical usage

from aeonic import init
from fastapi import FastAPI

app = FastAPI()

init({
    "api_key": "ag_test_123",
    "app": app,
    "service_name": "payments-api",
    "introspection_delay_ms": 2000,
    "capture_payloads": {
        "max_samples": 10,
    },
    # Optional – disable AgentGuard registration if needed:
    # "agentguard_enabled": False,
})

Accessing effective config (advanced)

For debugging or advanced inspection, you can read the effective configuration:

from aeonic.core import get_config

cfg = get_config()  # raises RuntimeError if init() was not called

---

aeonic.adapters.fastapi.agent_middleware

Purpose

FastAPI-compatible middleware (BaseHTTPMiddleware) that observes responses and captures samples only for requests that have been marked as agent routes via with_agent.

How to register

from fastapi import FastAPI
from aeonic.adapters.fastapi import agent_middleware

app = FastAPI()

app.add_middleware(agent_middleware)

Request/response handling

• For each request:
  • Calls the next handler (call_next(request)) first to let your business logic run.
  • Reads request.state.aeonic:
    • If missing or is_agent is falsy, returns the response immediately (non-agent route).
  • Skips assessment traffic:
    • If header x-aeonic-assessment: true is present, no samples are collected.
    • When the AEONIC_DEBUG environment variable is set, the middleware logs a debug message and includes optional x-aeonic-job-id for traceability.
  • Skips blocked/quarantined agents:
    • Uses should_collect_samples(agent_key) from aeonic.core.agent_status_cache.
    • agent_key is agent["name"] when available, otherwise "<METHOD>:<PATH>".
    • When AEONIC_DEBUG is set and an agent is blocked, a message is printed for easier debugging.

Payload capture

• Request payload:
  • Read from request.state.aeonic["req_payload"], which is set by the with_agent dependency (to avoid re-consuming the body).
  • Serialized via safe_serialize to guarantee JSON-safe structures.
• Response payload:
  • Attempts to read response.body directly when available (typical for JSON responses).
  • If only a body_iterator exists (streaming responses), it:
    • Iterates over the body iterator, collects all chunks, and rebuilds a new Response with the same status, headers, and media type.
  • Checks Content-Type header:
    • If it contains "json" and the body is non-empty, attempts json.loads() and then safe_serialize the result.
    • Any parsing failures are silently ignored (SDK never breaks your response).

Sample creation and buffering

• Builds a sample dictionary:
  • {"req": req_payload, "res": res_payload, "timestamp": int(time()), "status_code": status_code}.
• Reads SDK config via get_config():
  • Uses config["max_samples"], config["api_key"], and config["service_name"].
• If the status code is 2xx:
  • Resolves or creates a per-agent SampleBuffer in an in-memory _BUFFERS map, keyed by agent_key.
  • Caps the total number of in-memory buffers at 100 agents; when exceeded, the oldest key is evicted.
  • Calls buffer.add(sample):
    • When the buffer is full (max_samples), buffer.flush() is called and the payload is enqueued via enqueue(...) for asynchronous delivery to Aeonic.
• If the status code is non-2xx:
  • Immediately calls emit_error_sample(...) with a single-sample payload.

Key properties

• Only applies to routes where with_agent has attached metadata.
• Never throws or mutates your response in a way that changes semantics; at worst it falls back to no-op.
• Designed to be safe in high-throughput, async production environments.

---

aeonic.adapters.fastapi.with_agent(agent: AgentContext)

Purpose

FastAPI dependency factory that marks an endpoint as an agent route and attaches agent metadata and request payloads to the request.state.aeonic object.

Agent context (AgentContext)

AgentContext is a dict-like structure with keys such as:

• name (str, recommended): Logical agent name, used as the primary key for statistics and status.
• type (str | None): Business/domain category (e.g. "finance", "support", "claims").
• model (str | list[str] | None): Model identifier(s) powering the agent (e.g. "gpt-4o").
• Additional custom keys are supported and propagated as metadata.

Declaration-time behavior

• When you call with_agent(agent) at import time:
  • Immediately calls _register_agent_declaration(agent, framework="fastapi"), which in turn:
    • Resolves the current Aeonic config (if initialized) to obtain service_name.
    • Registers the agent in the global agent registry with:
      • name, type, model, framework="fastapi", and service_name.
      • route_path=None and http_method=None initially (filled in later by route introspection).
  • Stores a copy of the agent metadata (with "source": "manual") in an internal map keyed by the dependency function’s ID.
  • Also attaches metadata to the dependency function as __aeonic_metadata__ for robust detection, even when FastAPI wraps or decorates dependencies.

Request-time behavior

• with_agent(agent) returns an async dependency callable compatible with Depends:

from fastapi import Depends
from aeonic.adapters.fastapi import with_agent

@app.post(
    "/agent/finance",
    dependencies=[Depends(with_agent({"name": "RefundRiskAgent", "type": "finance", "model": ["gpt-4o"]}))],
)
async def finance_agent(payload: dict):
    ...

• On each request, the dependency:
  • Attempts to parse await request.json():
    • If parsing succeeds, body is serialized with safe_serialize.
    • On any error (non-JSON, invalid body, etc.), body is set to None.
  • Writes to request.state.aeonic:
    • {"is_agent": True, "agent": {**agent, "source": "manual"}, "req_payload": body}.
• This request.state.aeonic object is later consumed by agent_middleware to decide whether and how to capture samples.

Introspection helper: get_agent_metadata(func) (advanced)

• aeonic.adapters.fastapi.get_agent_metadata(func):
  • Attempts to recover an AgentContext from:
    • Direct ID lookup in the metadata map.
    • __aeonic_metadata__ attribute on the function or wrapped function.
    • Closure variables containing the agent context.
    • Underlying .call attribute (for FastAPI dependency objects).
  • Used internally by the Aeonic introspection system to match FastAPI routes and dependencies to their agents.
• Typical SDK consumers do not need to call this directly; it is useful only for advanced tooling or framework integrations built on top of Aeonic.

---

aeonic.adapters.django.agent_middleware

Purpose

Global Django middleware class that observes responses and captures samples only for views that have been marked as agent routes via the with_agent decorator.

How to register

# settings.py

MIDDLEWARE = [
    ...
    "aeonic.adapters.django.agent_middleware",
]

Request/response handling

• Constructed once per process with get_response.
• For each request:
  • Calls response = get_response(request) to run your Django view and middleware stack.
  • Reads request.aeonic:
    • If missing or is_agent is falsy, returns the response immediately (non-agent route).
  • Skips assessment traffic:
    • Uses request.META["HTTP_X_AEONIC_ASSESSMENT"] to detect the header X-Aeonic-Assessment: true.
    • When AEONIC_DEBUG is set, logs a message including optional HTTP_X_AEONIC_JOB_ID.
  • Skips blocked/quarantined agents:
    • Uses should_collect_samples(agent_key) with agent_key = agent["name"] or "<METHOD>:<PATH>".
    • When AEONIC_DEBUG is set and an agent is blocked, logs a message.

Payload capture

• Request payload:
  • Read from request.aeonic["req_payload"], which is set by the with_agent decorator.
  • Serialized via safe_serialize.
• Response payload:
  • Checks response["Content-Type"] header.
  • If it contains "json", attempts:
    • Decode response.content as UTF‑8.
    • Parse JSON and then safe_serialize the result.
  • Any parsing or decoding error is swallowed; the SDK never breaks your response pipeline.

Sample creation and buffering

• Builds a sample dictionary identical to the FastAPI middleware:
  • {"req": req_payload, "res": res_payload, "timestamp": int(time()), "status_code": status_code}.
• Reads SDK config via get_config() and uses max_samples, api_key, and service_name.
• For 2xx responses:
  • Uses a per-agent in-memory SampleBuffer in _BUFFERS, capped at 100 distinct agent keys with eviction of the oldest entry when full.
  • When a buffer fills up (max_samples), calls buffer.flush() and enqueues the batch via enqueue(...).
• For non-2xx responses:
  • Calls emit_error_sample(...) immediately with a single sample.

---

aeonic.adapters.django.with_agent(agent: AgentContext)

Purpose

Decorator that marks a Django view as an agent route and attaches agent metadata and payloads to request.aeonic.

Declaration-time behavior

• When the decorator is applied:
  • Calls _register_agent_declaration(agent, framework="django"):
    • Registers the agent in the global registry with name, type, model, framework="django", service_name, and route_path/http_method initially set to None.
  • Wraps the original view function with functools.wraps.
  • Stores a copy of the agent context (with "source": "manual") in an internal map keyed by the wrapped view’s ID.
  • Attaches metadata to the wrapped view as __aeonic_metadata__ for easier introspection later.

Request-time behavior

• The wrapped view:
  • Tries to build req_payload:
    • For POST, PUT, PATCH:
      • Reads CONTENT_TYPE from request.META.
      • If it contains "json" and request.body is non-empty:
        • Decodes body as UTF‑8, parses JSON, and safe_serializes the result.
      • If body exists but is not JSON:
        • Decodes as UTF‑8 string and serializes that.
    • For GET:
      • If request.GET has query parameters, serializes dict(request.GET).
      • If no query params, sets req_payload to an empty dict {}.
    • On any exception, req_payload falls back to None.
  • Sets:
    • request.aeonic = {"is_agent": True, "agent": {**agent, "source": "manual"}, "req_payload": req_payload}.
  • Calls and returns the original view function.

Introspection helper: get_agent_metadata(func) (advanced)

• aeonic.adapters.django.get_agent_metadata(func):
  • First checks the internal metadata map by ID.
  • Then falls back to the __aeonic_metadata__ attribute if present.
  • Used by internal route introspection to match Django URL patterns to their agents.
• Typical SDK users do not need to call this directly, but it is useful for custom tooling.

---

Summary of Public Entry Points

• Initialization
  • aeonic.init(config: dict) – one-time SDK setup, background introspection, and optional AgentGuard registration.
• FastAPI
  • aeonic.adapters.fastapi.agent_middleware – global middleware for observing and sampling agent routes.
  • aeonic.adapters.fastapi.with_agent(agent: AgentContext) – dependency factory to mark FastAPI endpoints as agent routes and capture request bodies.
• Django
  • aeonic.adapters.django.agent_middleware – middleware class for global observation and sampling of agent views.
  • aeonic.adapters.django.with_agent(agent: AgentContext) – decorator to mark Django views as agent routes and capture request data.
• Advanced (optional)
  • aeonic.core.get_config() – read the effective runtime SDK configuration.
  • aeonic.adapters.fastapi.get_agent_metadata(func) / aeonic.adapters.django.get_agent_metadata(func) – used by introspection to map framework objects back to agent metadata.

You can map each function/middleware above to your integration: use the descriptions as guidance and the code snippets as copy-paste examples.