Metadata-Version: 2.4
Name: aegislib
Version: 0.2.1
Summary: Aegis Python SDK -- policy-based tool guarding and LLM/agent observability (tracing, cost, redaction) for AI agents
Author-email: Aegis Team <bithal06@gmail.com>
Maintainer-email: mrsidrdx <bithal06@gmail.com>
License: MIT
Project-URL: Homepage, https://github.com/CloudMatos-ageis/aegis-python-sdk
Project-URL: Documentation, https://github.com/CloudMatos-ageis/aegis-python-sdk#readme
Project-URL: Repository, https://github.com/CloudMatos-ageis/aegis-python-sdk
Project-URL: Issues, https://github.com/CloudMatos-ageis/aegis-python-sdk/issues
Project-URL: Changelog, https://github.com/CloudMatos-ageis/aegis-python-sdk/blob/main/CHANGELOG.md
Keywords: ai,security,policy,agent,tool,guard,observability,tracing,opentelemetry,llm,genai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.25.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: backoff>=2.2.0
Requires-Dist: opentelemetry-api>=1.24.0
Provides-Extra: otel
Requires-Dist: opentelemetry-sdk>=1.24.0; extra == "otel"
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.24.0; extra == "otel"
Requires-Dist: opentelemetry-exporter-otlp-proto-grpc>=1.24.0; extra == "otel"
Provides-Extra: openai
Requires-Dist: openai>=1.0.0; extra == "openai"
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.40.0; extra == "anthropic"
Provides-Extra: bedrock
Requires-Dist: boto3>=1.34.0; extra == "bedrock"
Provides-Extra: google-genai
Requires-Dist: google-genai>=0.3.0; extra == "google-genai"
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.2.0; extra == "langchain"
Provides-Extra: langgraph
Requires-Dist: langgraph>=0.0.1; extra == "langgraph"
Provides-Extra: mcp
Requires-Dist: mcp>=1.0.0; extra == "mcp"
Provides-Extra: openai-agents
Requires-Dist: openai-agents>=0.0.1; extra == "openai-agents"
Provides-Extra: pydantic-ai
Requires-Dist: pydantic-ai-slim>=0.0.1; extra == "pydantic-ai"
Provides-Extra: claude-agent-sdk
Requires-Dist: claude-agent-sdk>=0.0.1; extra == "claude-agent-sdk"
Provides-Extra: mistral
Requires-Dist: mistralai>=1.0.0; extra == "mistral"
Provides-Extra: cohere
Requires-Dist: cohere>=5.0.0; extra == "cohere"
Provides-Extra: groq
Requires-Dist: groq>=0.4.0; extra == "groq"
Provides-Extra: together
Requires-Dist: together>=1.0.0; extra == "together"
Provides-Extra: pinecone
Requires-Dist: pinecone>=6.0.0; extra == "pinecone"
Provides-Extra: weaviate
Requires-Dist: weaviate-client>=4.23.0; extra == "weaviate"
Provides-Extra: qdrant
Requires-Dist: qdrant-client>=1.7.0; extra == "qdrant"
Requires-Dist: numpy<2.1.0; extra == "qdrant"
Provides-Extra: chroma
Requires-Dist: chromadb>=0.4.0; extra == "chroma"
Provides-Extra: crewai
Requires-Dist: crewai>=1.15.0; extra == "crewai"
Provides-Extra: autogen
Requires-Dist: autogen-agentchat>=0.4.0; extra == "autogen"
Requires-Dist: autogen-core>=0.4.0; extra == "autogen"
Provides-Extra: agno
Requires-Dist: agno>=2.0.0; extra == "agno"
Requires-Dist: openinference-instrumentation-agno>=1.0.0; extra == "agno"
Provides-Extra: haystack
Requires-Dist: haystack-ai>=2.0.0; extra == "haystack"
Provides-Extra: dspy
Requires-Dist: dspy>=3.0.0; extra == "dspy"
Provides-Extra: llama-index
Requires-Dist: llama-index-core>=0.11.0; extra == "llama-index"
Provides-Extra: instrumentations
Requires-Dist: aegislib[agno,anthropic,bedrock,chroma,claude-agent-sdk,cohere,crewai,dspy,google-genai,groq,haystack,langchain,langgraph,llama_index,mcp,mistral,openai,openai-agents,otel,pinecone,pydantic-ai,qdrant,together,weaviate]; extra == "instrumentations"
Provides-Extra: dev
Requires-Dist: pytest>=7.4.0; extra == "dev"
Requires-Dist: pytest-cov>=4.1.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: pytest-mock>=3.12.0; extra == "dev"
Requires-Dist: black>=23.0.0; extra == "dev"
Requires-Dist: ruff>=0.1.0; extra == "dev"
Requires-Dist: isort>=5.12.0; extra == "dev"
Requires-Dist: mypy>=1.7.0; extra == "dev"
Requires-Dist: build>=1.0.0; extra == "dev"
Requires-Dist: twine>=4.0.0; extra == "dev"
Provides-Extra: test
Requires-Dist: pytest>=7.4.0; extra == "test"
Requires-Dist: pytest-cov>=4.1.0; extra == "test"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "test"
Requires-Dist: pytest-mock>=3.12.0; extra == "test"
Requires-Dist: openai>=1.0.0; extra == "test"
Requires-Dist: opentelemetry-sdk>=1.24.0; extra == "test"
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.24.0; extra == "test"
Requires-Dist: opentelemetry-exporter-otlp-proto-grpc>=1.24.0; extra == "test"
Requires-Dist: langchain-core>=0.2.0; extra == "test"
Requires-Dist: langgraph>=0.0.1; extra == "test"
Requires-Dist: mcp>=1.0.0; extra == "test"
Requires-Dist: openai-agents>=0.0.1; extra == "test"
Requires-Dist: pydantic-ai-slim>=0.0.1; extra == "test"
Requires-Dist: claude-agent-sdk>=0.0.1; extra == "test"
Requires-Dist: mistralai>=1.0.0; extra == "test"
Requires-Dist: cohere>=5.0.0; extra == "test"
Requires-Dist: groq>=0.4.0; extra == "test"
Requires-Dist: together>=1.0.0; extra == "test"
Requires-Dist: pinecone>=6.0.0; extra == "test"
Requires-Dist: weaviate-client>=4.23.0; extra == "test"
Requires-Dist: qdrant-client>=1.7.0; extra == "test"
Requires-Dist: numpy<2.1.0; extra == "test"
Requires-Dist: chromadb>=0.4.0; extra == "test"
Requires-Dist: crewai>=1.15.0; extra == "test"
Requires-Dist: agno>=2.0.0; extra == "test"
Requires-Dist: openinference-instrumentation-agno>=1.0.0; extra == "test"
Requires-Dist: haystack-ai>=2.0.0; extra == "test"
Requires-Dist: dspy>=3.0.0; extra == "test"
Requires-Dist: llama-index-core>=0.11.0; extra == "test"
Requires-Dist: llama-index-llms-openai>=0.2.0; extra == "test"
Dynamic: license-file

# Aegis Python SDK

[![CI](https://github.com/CloudMatos-ageis/aegis-python-sdk/workflows/CI/badge.svg)](https://github.com/CloudMatos-ageis/aegis-python-sdk/actions)
[![PyPI version](https://badge.fury.io/py/aegislib.svg)](https://pypi.org/project/aegislib/)
[![Python versions](https://img.shields.io/pypi/pyversions/aegislib.svg)](https://pypi.org/project/aegislib/)
[![License](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)

**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](#installation)
- [Governance: guard a tool call](#governance-guard-a-tool-call)
- [Observability: trace an agent](#observability-trace-an-agent)
  - [Decorators](#decorators)
  - [Auto-instrumentation](#auto-instrumentation)
  - [Content capture and redaction](#content-capture-and-redaction)
  - [Distributed tracing and MCP](#distributed-tracing-and-mcp)
  - [Cost tracking and metrics](#cost-tracking-and-metrics)
  - [Scores and feedback](#scores-and-feedback)
  - [Bring your own backend (OTLP)](#bring-your-own-backend-otlp)
  - [Migrating from Respan or LangSmith](#migrating-from-respan-or-langsmith)
- [Configuration reference](#configuration-reference)
- [Governance deep dive](#governance-deep-dive)
  - [Approval workflow](#approval-workflow)
  - [Task management](#task-management)
  - [Error handling](#error-handling)
  - [Async support](#async-support)
- [API reference](#api-reference)
- [Framework integration examples](#framework-integration-examples)
- [Deployment topologies](#deployment-topologies)
- [Backward compatibility](#backward-compatibility)
- [Development](#development)

## Installation

```bash
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):

```bash
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

```python
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](#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:

```python
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:

```python
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:

```python
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>]`):

| Category | Providers/frameworks (`instruments=[...]` name) |
|---|---|
| LLM providers | `openai`, `azure_openai`, `anthropic`, `bedrock`, `google_genai`, `mistral`, `cohere`, `groq`, `together` |
| Agent frameworks | `langchain`, `langgraph`, `llama_index`, `openai_agents`, `pydantic_ai`, `claude_agent_sdk`, `crewai`, `autogen`, `agno`, `haystack`, `dspy` |
| Vector stores | `pinecone`, `weaviate`, `qdrant`, `chroma` |
| Protocol | `mcp` (client-side tracing, with duplicate-span suppression against a tool call an outer `@aegis_guard`/`@tool` span already owns) |

> `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):

```python
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:

```python
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:

```python
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:

```python
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):

```python
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

```python
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.

| `AegisConfig` field | Env var | Default | Notes |
|---|---|---|---|
| `base_url` | `AEGIS_BASE_URL` | `https://api.beta.aegissecurity.dev` | |
| `api_key` | `AEGIS_API_KEY` | -- (required) | |
| `timeout_s` | `AEGIS_TIMEOUT_S` | `10.0` | per-attempt timeout |
| `retries` | `AEGIS_RETRIES` | `2` | |
| `retry_max_time_s` | `AEGIS_RETRY_MAX_TIME_S` | `30.0` | total budget across all retries, independent of `timeout_s` |
| `user_agent` | -- | `aegis-python-sdk/<version>` | |
| `log_level` | `AEGIS_LOG_LEVEL` | `info` | |
| `debug` | `AEGIS_DEBUG` | `false` | enables console printers (`print_allow`, etc.); logging itself works regardless |
| `fail_mode` | `AEGIS_FAIL_MODE` | `closed` | `closed` blocks a tool call when the decision service is unreachable; `open` lets it through |
| `approval_polling_enabled` | `AEGIS_APPROVAL_POLLING_ENABLED` | `true` | |
| `approval_polling_initial_delay_s` | `AEGIS_APPROVAL_POLLING_INITIAL_DELAY_S` | `2.0` | |
| `approval_polling_max_delay_s` | `AEGIS_APPROVAL_POLLING_MAX_DELAY_S` | `60.0` | |
| `approval_polling_max_attempts` | `AEGIS_APPROVAL_POLLING_MAX_ATTEMPTS` | `50` | |
| `approval_polling_jitter_ratio` | `AEGIS_APPROVAL_POLLING_JITTER_RATIO` | `0.1` | |
| `approval_max_workers` | `AEGIS_APPROVAL_MAX_WORKERS` | `10` | bounded worker pool for background approval polling |
| `shutdown_flush_timeout_s` | `AEGIS_SHUTDOWN_FLUSH_TIMEOUT_S` | `30.0` | sets both `AegisConfig` and `ObservabilityConfig` |
| `install_signal_handlers` | `AEGIS_INSTALL_SIGNAL_HANDLERS` | `true` | chains to any handler you've already registered, never replaces it |
| `environment` / `service_name` / `service_version` / `tenant_id` | -- | -- | identity fields shared by governance and observability |
| `client_cert` / `client_key` / `ca_bundle` | -- | -- | mTLS |

| `ObservabilityConfig` field | Env var | Default | Notes |
|---|---|---|---|
| `enabled` | `AEGIS_OBSERVABILITY_ENABLED` | `false` | |
| `auto_instrument` | -- | `false` | |
| `instruments` | -- | `None` (all installed) | allow-list by name |
| `block_instruments` | -- | `set()` | deny-list, applied last (wins over an allow-list entry) |
| `sample_rate` | -- | -- | governance-relevant spans (deny/sanitize/approval_needed/violations) are always sent regardless of sampling |
| `content` | -- | `ContentPolicy()` | see [Content capture and redaction](#content-capture-and-redaction) |
| `exporters` | -- | `[native]` | see [Bring your own backend](#bring-your-own-backend-otlp) |
| `metrics_enabled` / `metrics_export_interval_s` | -- | `true` / `60.0` | |
| `stream_token_events_enabled` / `stream_token_event_sample_rate` | -- | `false` / -- | opt-in `aegis.llm.token` per-chunk events, sampled independently of trace sampling |
| `batch_max_spans` / `batch_max_bytes` / `batch_flush_interval_s` | -- | -- | adapted automatically to the backend's advertised limits via a capability handshake |
| `buffer_max_spans` / `export_timeout_s` | -- | -- | bounded buffer -- see NFR-2 in the PRD |

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:

```python
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:

```python
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

```python
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:

```python
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:

```python
@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](#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

```python
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:

```python
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:

```python
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

```python
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](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

```bash
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](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](LICENSE).
