Aegis Python SDK

Two independently-adoptable interfaces for AI agents, over one shared core:

• aegis.governance -- policy-based tool guarding. Wrap any tool call with @aegis_guard and get real-time allow/deny/sanitize/approval decisions enforced before the call runs.
• aegis.observability -- LLM/agent tracing. Decorate or auto-instrument an agent app and get correctly-parented, sampled spans (cost, tokens, prompts, tool calls, retries) exported to your own backend or Aegis's, with built-in PII redaction and OpenTelemetry GenAI interop.

Neither requires the other. Install one, both, or add the second later without touching your existing integration -- every 0.1.x governance call keeps working unchanged in 0.2.x.

Table of contents

• Installation
• Governance: guard a tool call
• Observability: trace an agent
  • Decorators
  • Auto-instrumentation
  • Content capture and redaction
  • Distributed tracing and MCP
  • Cost tracking and metrics
  • Scores and feedback
  • Bring your own backend (OTLP)
  • Migrating from Respan or LangSmith
• Configuration reference
• Governance deep dive
  • Approval workflow
  • Task management
  • Error handling
  • Async support
• API reference
• Framework integration examples
• Deployment topologies
• Backward compatibility
• Development

Installation

pip install aegislib

Auto-instrumentation for a specific provider or framework is an optional extra -- none of these are pulled in by the core install, so an air-gapped or dependency-conscious deployment stays small (httpx, pydantic, backoff, opentelemetry-api are the only core dependencies):

pip install aegislib[openai]              # one provider
pip install aegislib[openai,langchain]    # a few
pip install aegislib[instrumentations]    # everything except autogen (see note below)
pip install aegislib[otel]                # OTLP export (opentelemetry-sdk + exporters)

import aegis never makes a network call and never talks to any of these libraries unless you enable auto-instrumentation for them, so installing the SDK never has side effects.

Governance: guard a tool call

from aegis import AegisConfig, DecisionClient, aegis_guard

cfg = AegisConfig(api_key="your-api-key")
client = DecisionClient(cfg)

@aegis_guard(client, agent_id="ops-agent", tool_name="slack.post_message")
def post_to_slack(channel: str, text: str) -> str:
    print(f"Sending message: {text} to {channel}")
    return "sent"

post_to_slack("#support", "Hello!")

# A tool guarded with a policy that requires approval doesn't run inline -- it's queued,
# and runs in the background once approved (see "Approval workflow" below).
@aegis_guard(client, agent_id="admin-agent", tool_name="database.drop_table")
def drop_database_table(table_name: str) -> str:
    return f"Dropped table: {table_name}"

drop_database_table("users")  # returns immediately; executes later if/when approved

A denied call raises aegis.ForbiddenError -- it never silently returns a string that looks like a successful result (see Backward compatibility for why this changed in 0.2.0 if you're upgrading from 0.1.x).

@aegis_guard is also where governance and observability actually join: it takes keyword-only trace (default True -- every guarded call becomes an observability span automatically, with no separate @tool needed, if observability is enabled), capture_args/capture_result (override the global content-capture setting for just this tool), fail_mode (override AegisConfig.fail_mode per tool -- e.g. one specific low-risk tool can fail open while everything else stays closed), and span_attributes. None of this does anything if observability is off; it's inert overhead-free plumbing until you turn it on.

Observability: trace an agent

Observability is off by default (AEGIS_OBSERVABILITY_ENABLED=false) -- constructing Aegis() with nothing turned on starts no background thread and sends nothing. Turn it on explicitly:

from aegis import Aegis

aegis = Aegis(
    api_key="your-api-key",
    environment="production",
    service_name="checkout-agent",
    auto_instrument=True,   # patch every installed, supported provider/framework automatically
)
# ... your agent code runs, spans are created and buffered ...
aegis.shutdown()            # flushes the buffer with a bounded timeout; also runs at process exit

Aegis() is idempotent per process: calling it again while the first instance is alive returns the same instance (with a warning) instead of building a second exporter thread and re-instrumenting every client a second time.

Decorators

Six span types cover an agent's real structure -- @workflow (a whole run), @agent (one agent's turn), @task, @tool, @plan, and @retrieval. Every decorator works on a sync function, an async function, a generator, and an async generator:

from aegis import workflow, agent, tool

@workflow(name="handle-support-ticket")
def handle_ticket(ticket_id: str) -> str:
    return run_agent(ticket_id)

@agent(name="support-agent")
def run_agent(ticket_id: str) -> str:
    return look_up_order(ticket_id)

@tool(name="orders.lookup", span_attributes={"system": "orders-db"})
def look_up_order(ticket_id: str) -> str:
    return "order-status: shipped"

A tool guarded with @aegis_guard and traced with @tool in the same call gets a governance extension on its span (the decision effect, policy id, and any violations) -- one span joins both interfaces for that call, which is exactly what makes an Aegis trace different from a plain OTel trace: you see why a step was allowed or blocked, not just that it happened.

Auto-instrumentation

auto_instrument=True patches whatever's actually installed -- nothing errors if a library isn't present, it's just skipped. Narrow or widen the set explicitly:

aegis = Aegis(api_key="...", auto_instrument=True, instruments=["openai", "anthropic"])
aegis = Aegis(api_key="...", auto_instrument=True, block_instruments={"crewai"})

25 instrumentations ship in-tree, each behind its own extra (pip install aegislib[<name>]):

aegislib[instrumentations] bundles all of the above except autogen -- autogen-core's protobuf pin is unsatisfiable alongside chroma/qdrant in the same environment. Install aegislib[autogen] on its own if you need it.

Two instrumentations patch via a custom OTel SpanProcessor bridge instead of monkeypatching a client directly (agno, autogen, pydantic_ai) -- transparent to you, just worth knowing if you're also running your own OTel SpanProcessor for the same library.

Content capture and redaction

Prompt/response/tool-argument content is opt-in, off by default, and topology-auto-detected (a self-hosted on-prem deployment defaults more conservatively than a SaaS one talking to Aegis's own backend):

from aegis import Aegis

aegis = Aegis(
    api_key="...",
    capture_mode="metadata_only",   # "none" | "metadata_only" | "full"
    redact_pii=True,                # on by default when capture is enabled
)

Built-in detectors catch email, phone, credit card, IBAN, SSN, PAN, Aadhaar, AWS keys, PEM blocks, JWTs, and bearer tokens -- redaction runs in-process, before anything reaches the export buffer, so an unredacted value never gets serialized or leaves the process. Add your own regex patterns, a field-name deny-list, or a custom hook:

from aegis import ObservabilityConfig, ContentPolicy

policy = ContentPolicy(
    capture_mode="full",
    redaction_patterns=[r"\bINTERNAL-\d{6}\b"],
    redact_field_names={"ssn", "internal_notes"},
    hook=lambda value: my_company_redactor(value),
)

A redaction-hook failure is fail-closed: the field is dropped and a counter increments (visible via aegis.get_telemetry_stats()), it never leaks the unredacted value. Large fields (over a configurable size) are offloaded to /v1/telemetry/content rather than inlined on the span.

Sanitized values (from a governance sanitize decision) are never recorded in telemetry -- only the sanitized field names are, since the whole point of sanitizing a value is that the original was unsafe to keep around.

Distributed tracing and MCP

W3C traceparent/tracestate encode/decode, extraction, and injection are built in, so a trace started in one process correctly continues in another:

from aegis import extract_traceparent, inject_traceparent, remote_trace_context, tool

# Process A: injecting only does something inside an active span -- it returns a NEW
# headers dict (never mutates the one you pass in), built from the current span's ids.
@tool(name="call-downstream-service")
def call_downstream():
    headers = inject_traceparent({"Content-Type": "application/json"})
    requests.post(url, headers=headers)

# Process B: continue the trace that arrived over the wire
ctx = extract_traceparent(incoming_headers)  # None if the header was missing/malformed
if ctx is not None:
    with remote_trace_context(ctx):
        handle_request()
else:
    handle_request()

MCP has no top-level headers, so the same propagation also works over params._meta (inject_traceparent_into_mcp_meta / extract_traceparent_from_mcp_meta) -- the mcp instrumentation uses this automatically for client-side spans.

Cost tracking and metrics

Every LLM span gets an automatic cost estimate from an embedded model-pricing catalog (token usage x the model's per-token rate), attached as aegis.cost.* attributes -- no configuration needed. estimate_cost() and get_catalog_entry() are available directly if you want to price something outside a span. Aegis-specific metrics (span counts, export latency, buffer state) ship with cardinality enforcement so a runaway attribute value can't blow up your metrics backend's cost.

Scores and feedback

Attach a quality score or human feedback to a trace after the fact:

from aegis import score, ascore

score(trace_id="...", name="helpfulness", value=0.9, comment="Resolved on first reply")
await ascore(trace_id="...", name="thumbs_up", value=1.0)

Bring your own backend (OTLP)

Export to Aegis's own collector, your own OTLP-compatible backend, or both at once (dual-write, useful during a migration):

from aegis import Aegis, ExporterConfig

aegis = Aegis(
    api_key="...",
    exporters=[
        ExporterConfig(kind="aegis_native"),                                    # Aegis's own backend
        ExporterConfig(kind="otlp_grpc", endpoint="https://otel.internal:4317"),  # your own, via gRPC
        # or ExporterConfig(kind="otlp_http", endpoint="https://otel.internal:4318")
    ],
)

kind is an open string, not a fixed enum, so a new exporter kind never needs an SDK release to use -- aegis_native, otlp_grpc, and otlp_http are simply the three this SDK ships support for out of the box.

gen_ai.* attribute keys follow the OpenTelemetry GenAI semantic conventions exactly (pinned to a specific upstream commit, since that spec is still under active development); anything with no OTel equivalent -- decision effect, policy id, approval state, cost, tenant -- lives under its own aegis.* namespace rather than being force-fit into gen_ai.*.

Migrating from Respan or LangSmith

from aegis import traceable, translate_legacy_fields

@traceable(run_type="chain")   # maps LangSmith's run_type onto an Aegis span type
def my_chain(input: str) -> str:
    ...

# Opt-in field-name translation -- never automatic, so it can't silently reinterpret
# a field name your own code already uses for something else.
normalized = translate_legacy_fields(my_legacy_payload)

Configuration reference

AegisConfig covers governance + transport; ObservabilityConfig (nested at AegisConfig.observability, or passed as flat kwargs to Aegis(...)) covers tracing. Every field has an environment variable; an explicit constructor argument always wins over its env var.

The full environment-variable table (every field, every default) is in docs/governance-observability-prd, Appendix A.

Governance deep dive

Approval workflow

A tool guarded by a policy that requires approval doesn't block -- it's queued and polled in the background, then executed once approved:

from aegis import AegisConfig, DecisionClient, aegis_guard

cfg = AegisConfig(
    api_key="your-api-key",
    approval_polling_enabled=True,
    approval_polling_initial_delay_s=2.0,
    approval_polling_max_delay_s=60.0,
    approval_polling_max_attempts=50,
)
client = DecisionClient(cfg)

def on_approval_complete(result, error):
    if error:
        print(f"Approval failed: {error}")
    else:
        print(f"Approved and executed: {result}")

@aegis_guard(client, agent_id="admin-agent", tool_name="database.drop_table", approval_callback=on_approval_complete)
def drop_database_table(table_name: str) -> str:
    return f"Dropped table: {table_name}"

drop_database_table("users")  # returns immediately; on_approval_complete fires later

Task management

Every approval-required call is tracked as a task you can query:

from aegis import get_global_task_manager

task_manager = get_global_task_manager()

pending = task_manager.list_tasks(status="pending")
agent_tasks = task_manager.list_tasks(agent_id="admin-agent")

task = task_manager.get_task("task-uuid-here")
if task:
    print(f"Task status: {task.status}, created: {task.created_at}")

task_manager.clear_completed(max_age_seconds=3600)

stats = task_manager.get_stats()
print(f"Total tasks: {stats['total']}, Pending: {stats['pending']}")

TaskManager is bounded by default (max_tasks=10_000, ttl_s=3600) -- it sweeps expired tasks and evicts down to the max size on every create_task() call, so a long-running process doesn't need to remember to call clear_completed() on its own schedule. Pass 0 to either bound to disable it.

Decision effects: allow (proceeds normally), deny (raises ForbiddenError), sanitize (parameters modified before execution), approval_needed (queued as above).

Error handling

from aegis import ForbiddenError, AuthError

@aegis_guard(client, agent_id="agent", tool_name="risky_tool")
def risky_operation(data: str) -> str:
    return f"Processing: {data}"

try:
    result = risky_operation("sensitive data")
except ForbiddenError as e:
    print(f"Operation blocked: {e}")
except AuthError as e:
    print(f"Authentication failed: {e}")

The full hierarchy: AegisError is the base every SDK exception inherits from. AuthError, ForbiddenError, BadRequestError, and NotFoundError are re-exported from aegis directly (from aegis import ForbiddenError). TransportError, TimeoutError, and PolicyEvalError live in aegis.errors (from aegis.errors import TransportError) rather than the top-level package -- TimeoutError deliberately isn't re-exported at the top level so it doesn't shadow Python's builtin exception of the same name in a caller's from aegis import *-style code. Catching AegisError covers all seven.

Pass additional context with a request:

session_data = {"user_id": "user-123", "session_id": "sess-456", "context": "production"}
response = client.decide(agent_id="agent", tool_name="tool", params={"key": "value"}, session=session_data)

Async support

Every decorator and client entry point supports sync functions, async functions, generators, and async generators. The async governance path uses a genuinely async HTTP request under the hood (DecisionClient.adecide()), not a sync call wrapped in a thread -- guarding an async tool call never blocks the event loop:

@aegis_guard(client, agent_id="agent", tool_name="async_tool")
async def async_operation(data: str) -> str:
    await some_async_work(data)
    return "done"

result = await async_operation("data")

API reference

Governance: AegisConfig, DecisionClient (decide, adecide, get_decision_status, close, aclose), DecisionClient.guarded() (context manager for dynamic tool dispatch), aegis_guard, Decision, DecisionRequest, DecisionResponse, ToolCall.

Approval workflow: ApprovalExecutor (start, shutdown, submit_for_approval, get_queue_size, get_pending_tasks), ApprovalTask, ApprovalTaskQueue, get_global_executor / set_global_executor / shutdown_global_executor.

Task management: TaskManager (create_task, update_status, get_task, list_tasks, delete_task, clear_completed, get_stats), TaskInfo, TaskStatus, get_global_task_manager / reset_global_task_manager.

Observability facade: Aegis (top-level entry point; idempotent per process, shutdown()).

Observability decorators: workflow, agent, task, tool, plan, retrieval -- each takes an optional name, plus span_attributes, metadata, tags keyword arguments.

Observability core: Span, SpanType, SpanStatus, AegisTracer (get_tracer()), AegisNativeExporter, get_current_trace_id, update_current_span, add_event, record_exception, span_attributes, propagate_attributes.

Distributed tracing: RemoteTraceContext, remote_trace_context, extract_traceparent, format_traceparent, inject_traceparent, extract_traceparent_from_mcp_meta, inject_traceparent_into_mcp_meta.

Instrumentation: Instrumentation (protocol for writing your own), plus one class per provider/framework listed in Auto-instrumentation (e.g. OpenAIInstrumentation, LangChainInstrumentation, ... 25 total) for explicit registration instead of auto_instrument=True.

Cost & scores: PricingEntry, estimate_cost, get_catalog_entry, score, ascore.

Migration: traceable, translate_legacy_fields, LEGACY_FIELD_ALIASES.

Capability handshake: CapabilityClient, CapabilitiesResponse, get_global_capability_client, reset_global_capability_clients.

Config: AegisConfig, ObservabilityConfig, ContentPolicy, ExporterConfig.

Errors: AegisError, AuthError, ForbiddenError, BadRequestError, NotFoundError (top level); TransportError, TimeoutError, PolicyEvalError (aegis.errors).

Framework integration examples

FastAPI services

from fastapi import FastAPI
from aegis import AegisConfig, DecisionClient, aegis_guard

app = FastAPI()
client = DecisionClient(AegisConfig(api_key="your-api-key"))

@aegis_guard(client, agent_id="api-agent", tool_name="user.delete")
async def delete_user(user_id: str) -> dict:
    return {"deleted": user_id}

@app.delete("/users/{user_id}")
async def delete_user_endpoint(user_id: str):
    return await delete_user(user_id)

LangChain / LangGraph agents

Auto-instrumentation is usually enough -- no code changes inside the chain/graph itself:

from aegis import Aegis

aegis = Aegis(api_key="your-api-key", auto_instrument=True, instruments=["langchain", "langgraph"])
# build and run your LangChain/LangGraph agent as normal; spans are emitted automatically

Guard an individual tool the agent calls the same way as anywhere else:

from aegis import aegis_guard
from langchain.tools import Tool

@aegis_guard(client, agent_id="langchain-agent", tool_name="web_search")
def search_web(query: str) -> str:
    return f"Results for: {query}"

search_tool = Tool(name="search", func=search_web, description="Search the web")

Async applications

import asyncio
from aegis import AegisConfig, DecisionClient, aegis_guard

client = DecisionClient(AegisConfig(api_key="your-api-key"))

@aegis_guard(client, agent_id="async-agent", tool_name="fetch_data")
async def fetch_data(url: str) -> dict:
    return {"url": url, "data": "..."}

async def main():
    result = await fetch_data("https://api.example.com/data")
    print(result)

asyncio.run(main())

Deployment topologies

The same SDK config adapts to where your backend actually lives, auto-detected from base_url (explicit topology/capture_mode overrides always win):

• SaaS -- talking to Aegis's own hosted backend.
• On-premises / VPC -- your own aegis-data-plane deployment; storage-profile aware ("Compact" vs "Scale") via the capability handshake.
• Bring-your-own-backend -- OTLP only, no Aegis backend involved at all.
• Dual-write -- both at once, for migrating off an existing OTel setup without a hard cutover.

Backward compatibility

Every 0.1.4 import path, class, function signature, and environment variable keeps working unchanged in 0.2.x -- upgrading without touching your config gets you the defect fixes below plus an inert aegis.observability (off by default) and nothing else. The two exceptions, both intentional and both already the unconditional behavior as of 0.2.0:

• @aegis_guard on deny always raises ForbiddenError. It used to return a formatted string instead, which looked like a successful result to any caller -- in an agent loop, that string got fed back to the model as tool output, and the agent never learned the call was denied.
• Explicit AegisConfig constructor arguments always win over environment variables. Previously an environment variable silently overrode an explicit argument of the same name.

See CHANGELOG.md for the complete list of defect fixes bundled into 0.2.0 (connection pooling, retry handling, bounded task/thread growth, signal-safe shutdown, and more) and the full "Added" list for aegis.observability.

Development

git clone https://github.com/CloudMatos-ageis/aegis-python-sdk.git
cd aegis-python-sdk
pip install -e ".[dev]"
pytest                  # unit tests, coverage-gated (see CLAUDE.md's NFR table)
mypy aegis               # strict mode
ruff check .

See CONTRIBUTING.md for the full workflow, and docs/governance-observability-prd for the design rationale behind aegis.observability if you're touching that subsystem.

License

MIT -- see LICENSE.