aeonic-agentguard-sdk-python 0.1.2

Official Python SDK for Aeonic — runtime monitoring and governance for AI agents

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

Requirement	Details
Python	>=3.9
Core dependency	requests>=2.25.0 (installed automatically)
FastAPI (optional)	fastapi>=0.68.0 — install via [fastapi] extra
Django (optional)	django>=3.2.0 — install via [django] extra

Environment variables (optional)

Variable	Purpose
AEONIC_ENDPOINT	Aeonic platform endpoint URL (provided in your Aeonic account or onboarding)
AEONIC_DEBUG	Set to any value to enable debug logging
ORIGIN / BASE_URL	Your application's public URL, sent with sample payloads
PORT / HOST	Used to auto-detect your app origin when ORIGIN is not set

---

🚀 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

Feature	Behavior
Only marked routes tracked	✅
Async safe	✅
Middleware-based	✅
No contextvars hacks	✅
Production ready	✅

---

🏛️ 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

Feature	Behavior
Declarative agent marking	✅
Middleware-based capture	✅
Sync-safe	✅
No framework coupling	✅
Production ready	✅

---

🧠 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

---

🧩 Advanced Use Cases

• Background workers
• Batch inference jobs
• Non-HTTP agents

(Contact the Aeonic team for advanced SDK APIs.)

---

🔧 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.