Metadata-Version: 2.4
Name: enterprise-agentic-ai-framework
Version: 0.23.0
Summary: Enterprise Agentic AI Framework SDK
License-Expression: Apache-2.0
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2.0
Provides-Extra: authentication
Requires-Dist: python-keycloak>=4.0; extra == 'authentication'
Provides-Extra: dev
Requires-Dist: httpx2>=2.0; extra == 'dev'
Requires-Dist: hvac>=2.0; extra == 'dev'
Requires-Dist: mcp>=2.0; extra == 'dev'
Requires-Dist: opentelemetry-api>=1.27; extra == 'dev'
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.27; extra == 'dev'
Requires-Dist: opentelemetry-exporter-prometheus>=0.48b0; extra == 'dev'
Requires-Dist: opentelemetry-sdk>=1.27; extra == 'dev'
Requires-Dist: prometheus-client>=0.20; extra == 'dev'
Requires-Dist: psycopg[binary]>=3.1; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: python-keycloak>=4.0; extra == 'dev'
Requires-Dist: qdrant-client>=1.9; extra == 'dev'
Requires-Dist: redis>=5.0; extra == 'dev'
Provides-Extra: mcp-client
Requires-Dist: httpx2>=2.0; extra == 'mcp-client'
Requires-Dist: mcp>=2.0; extra == 'mcp-client'
Provides-Extra: memory
Requires-Dist: psycopg[binary]>=3.1; extra == 'memory'
Requires-Dist: qdrant-client>=1.9; extra == 'memory'
Requires-Dist: redis>=5.0; extra == 'memory'
Provides-Extra: metrics
Requires-Dist: opentelemetry-api>=1.27; extra == 'metrics'
Requires-Dist: opentelemetry-exporter-prometheus>=0.48b0; extra == 'metrics'
Requires-Dist: opentelemetry-sdk>=1.27; extra == 'metrics'
Requires-Dist: prometheus-client>=0.20; extra == 'metrics'
Provides-Extra: observability
Requires-Dist: opentelemetry-api>=1.27; extra == 'observability'
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.27; extra == 'observability'
Requires-Dist: opentelemetry-exporter-prometheus>=0.48b0; extra == 'observability'
Requires-Dist: opentelemetry-sdk>=1.27; extra == 'observability'
Requires-Dist: prometheus-client>=0.20; extra == 'observability'
Provides-Extra: postgres
Requires-Dist: psycopg[binary]>=3.1; extra == 'postgres'
Provides-Extra: qdrant
Requires-Dist: qdrant-client>=1.9; extra == 'qdrant'
Provides-Extra: redis
Requires-Dist: redis>=5.0; extra == 'redis'
Provides-Extra: tracing
Requires-Dist: opentelemetry-api>=1.27; extra == 'tracing'
Requires-Dist: opentelemetry-exporter-otlp-proto-http>=1.27; extra == 'tracing'
Requires-Dist: opentelemetry-sdk>=1.27; extra == 'tracing'
Provides-Extra: vault
Requires-Dist: hvac>=2.0; extra == 'vault'
Description-Content-Type: text/markdown

# enterprise-agentic-ai-framework

An enterprise governance framework for building single- and multi-agent
AI systems in Python: authorization, guardrails, observability, secrets
management, LLM gateway access, and a full production evaluation
suite, all as one consistent stack instead of one-off code per project.

```bash
pip install enterprise-agentic-ai-framework
```

The import name is `agentic_ai` (the PyPI distribution name is longer
for naming reasons, the package you actually `import` is not):

```python
from agentic_ai.gateway import LiteLLMGateway
```

**Full step-by-step documentation** (prerequisites, mandatory
parameters, and the complete request-flow architecture) lives in
[`docs/`](docs/README.md) — start at
[`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md).

## Status

This is an early release. **The LLM gateway, the full evaluation suite,
Memory & State, Context Engineering, Secrets Management, Guardrails &
Content Safety, Authentication, Identity, Governance (in full -
authorization, audit, security, data, compliance, responsible AI,
finops), Resilience & Reliability, Observability, Monitoring &
Alerting, and Core Agent Capabilities are implemented today** -
everything else below is scaffolded (the module exists, it's empty)
and not yet usable. This table will be kept current as modules land,
not written once and left stale.

| Module | Status |
|---|:---:|
| `gateway` - LLM gateway (LiteLLM proxy client) | ✅ Implemented |
| `evaluation` - Agent/LLM/Tools/Multi-Agent/RAG/Security/Platform/Memory/Drift evaluation (48 metrics, see below) | ✅ Implemented |
| `memory` - session/agent/short-term/working/long-term/semantic/episodic/procedural/document/shared memory & state (see below) | ✅ Implemented |
| `context` - context engineering: assembly, write, select, compress, isolate (see below) | ✅ Implemented |
| `secrets` - secrets management: HashiCorp Vault (KV v2, dynamic secrets, Transit encryption) (see below) | ✅ Implemented |
| `guardrails` - input/output guardrails, injection/jailbreak/content-safety/PII detection, tool gating, rate limiting, human-in-the-loop (see below) | ✅ Implemented |
| `authentication` - credential mechanics: Keycloak (OIDC/OAuth2 grants, MFA, mTLS, token validation/refresh/revocation, auto-refresh) (see below) | ✅ Implemented |
| `identity` - claims/data layer: TokenClaims, UserInfo, Tenant Identity Isolation (see below) | ✅ Implemented |
| `governance.authorization` - PDP/PEP via Open Policy Agent: RBAC/ABAC/fine-grained authorization (see below) | ✅ Implemented |
| `governance.audit` - durable, hash-chained audit trail + adapters for every other module's audit wrapper (see below) | ✅ Implemented |
| `governance.security` - agent/multi-agent/memory/RAG/privacy security, kill switch, SIEM export, red-team self-test (see below) | ✅ Implemented |
| `governance.data` - Data Classification, Access Governance, Privacy/PII, Encryption, Isolation, DLP, Retention, Lineage, Quality, Lifecycle, RAG Data Governance, AI Data Access Control (see below) | ✅ Implemented |
| `governance.compliance` - general-purpose requirement/evidence register (see below) | ✅ Implemented |
| `governance.responsibleai` - AI Risk Management, Human Oversight & Accountability, Transparency & Explainability, Fairness & Bias, AI Safety & Reliability, AI Governance & Compliance (see below) | ✅ Implemented |
| `governance.finops` - Cost Monitoring & Visibility, Cost Allocation & Attribution, Budget & Cost Controls, LLM/Model Cost Optimization, Resource Optimization, Cost Forecasting & Reporting (see below) | ✅ Implemented |
| `resilience` - High Availability, Fault Tolerance, Retry & Timeout Management, Circuit Breaker & Bulkhead, Fallback & Recovery, State Checkpointing & Resume, Disaster Recovery, Capacity & Load Management (see below) | ✅ Implemented |
| `observability` - Distributed Tracing (OpenTelemetry -> Jaeger), Application & Infrastructure Metrics (OpenTelemetry -> Prometheus), LLM/Agent/RAG/Tool Observability, Logs & Events (Postgres), Alerting & Anomaly Detection, Correlation & RCA, Grafana dashboard provisioning (see below) | ✅ Implemented |
| `monitoring` - Health Monitoring, Performance Monitoring, AI/Agent Monitoring, Quality Monitoring, Cost Monitoring, Security/Guardrail Monitoring, Alerting & Incident Management (PagerDuty + Grafana dashboard) (see below) | ✅ Implemented |
| `core` - Goal/Plan/Reasoning/Tool Selection/Task Execution/State Management/Memory Access/Observation/Error Recovery/Goal Completion, async AgentExecutor with checkpoint+resume, self-learning knowledge base, Router/Hierarchical/Sequential/Parallel/Graph multi-agent patterns (see below) | ✅ Implemented |
| `code_interpreter` (`agentic_ai.core.code_interpreter`) - read/write/edit/grep/glob/run-any-command tool, sandboxed (Docker/local-jail/Python) (see below) | ✅ Implemented |
| `mcp_client` (`agentic_ai.mcp_client`) - MCP (Model Context Protocol) client: local (stdio) and remote (Streamable HTTP, optional OAuth 2.1) server connections, multi-server aggregation, progressive tool discovery, full AgentExecutor integration (see below) | ✅ Implemented |

**A naming note, not a contradiction**: `evaluation.memory` and the
top-level `memory` module are different things. `evaluation.memory`
*measures* something (was a memory retrieval accurate/consistent?) from
data you already collected. The top-level `memory` module described
below *is* the live storage layer - the thing `evaluation.memory` would
be measuring. Same relationship for `evaluation.security` (still
planned as a top-level module) vs. the eventual live `security`
enforcement layer.

## Prerequisites

**This library is a client, not a server.** Before any of the examples
below will work, you need a LiteLLM proxy already running somewhere
reachable - `agentic_ai.gateway` never installs, starts, stops, or
otherwise manages that process for you. Set it up once:

**1. Install LiteLLM's proxy** (a separate package from this library):

```bash
pip install 'litellm[proxy]'
```

**2. Register at least one model.** Create `litellm_config.yaml` -
this example routes the model name `gpt-4o-mini` to OpenAI, reading the
real provider key from an environment variable (never hardcode it in
the YAML):

```yaml
model_list:
  - model_name: gpt-4o-mini
    litellm_params:
      model: openai/gpt-4o-mini
      api_key: os.environ/OPENAI_API_KEY
```

Any provider LiteLLM supports works the same way - Anthropic, Azure
OpenAI, Bedrock, a local Ollama model, etc.; only `litellm_params`
changes. See LiteLLM's own docs for the full provider list.

**3. Set the real provider key and start the proxy:**

```bash
export OPENAI_API_KEY=sk-...
litellm --config litellm_config.yaml --port 4000
```

**4. Confirm it's actually up** before writing any Python against it:

```bash
curl http://localhost:4000/health/liveliness
# -> "I'm alive!"
```

If that curl fails, nothing below will work either - fix connectivity
to the proxy first; `agentic_ai.gateway`'s errors will otherwise (correctly)
just tell you the same thing: it can't reach `http://localhost:4000`.

Only once you have a real, running, reachable LiteLLM proxy do the
examples below have anything to talk to.

## Quickstart: LLM Gateway

### 1. Connect to it

```python
from agentic_ai.gateway import LiteLLMGateway

# No arguments needed for the common case: connects to
# http://localhost:4000, LiteLLM's own default port.
gateway = LiteLLMGateway()

reply = gateway.complete(
    model="gpt-4o-mini",  # must be registered on your proxy, e.g. in litellm_config.yaml
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Name three benefits of distributed tracing."},
    ],
)
print(reply)
```

### 2. Configuring host, port, and auth

```python
from agentic_ai.gateway import LiteLLMGateway

# Custom port - your proxy isn't on LiteLLM's default 4000
gateway = LiteLLMGateway(port=5001)

# Custom host and port - a proxy running elsewhere on your network
gateway = LiteLLMGateway(host="litellm.internal", port=8080)

# Full base_url - anything host/port can't express (TLS, a path prefix)
gateway = LiteLLMGateway(base_url="https://litellm.example.com/proxy")

# A proxy that requires a virtual key
gateway = LiteLLMGateway(api_key="sk-...")  # resolve this from your own
                                             # secrets store - the gateway
                                             # module doesn't fetch it for you
```

### 3. The full response, not just the text

`complete()` is a convenience wrapper around `chat_completion()`, which
returns the full OpenAI-compatible response body (usage, finish_reason,
etc.) when you need more than just the message content:

```python
result = gateway.chat_completion(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Summarize this in one sentence: ..."}],
    temperature=0.2,
    max_tokens=200,
)
print(result["choices"][0]["message"]["content"])
print(result["usage"])
```

### 4. Handling errors

The gateway never lets a raw network exception escape - callers get one
of two exceptions, so "the proxy is down" and "the proxy rejected the
request" are never conflated:

```python
from agentic_ai.gateway import GatewayConnectionError, GatewayRequestError, LiteLLMGateway

gateway = LiteLLMGateway()

try:
    reply = gateway.complete("gpt-4o-mini", [{"role": "user", "content": "hi"}])
except GatewayConnectionError:
    # Nothing is listening at gateway.base_url at all - is LiteLLM
    # actually running? (see Prerequisites above)
    ...
except GatewayRequestError as e:
    # The proxy responded, but with an error (bad model name, missing
    # api_key, malformed request) - e includes the proxy's own message.
    print(e)
```

### 5. Cleaning up

`LiteLLMGateway` holds an open HTTP connection pool; close it when
you're done, or use it as a context manager:

```python
with LiteLLMGateway() as gateway:
    reply = gateway.complete("gpt-4o-mini", [{"role": "user", "content": "hi"}])
# connection pool closed automatically here
```

### 6. LLM Audit

```python
from agentic_ai.gateway.wrappers import AuditedLiteLLMGateway

audited = AuditedLiteLLMGateway(gateway, sink=lambda e: logger.info("llm", extra=e.model_dump()))
audited.chat_completion("gpt-4o-mini", messages)  # event carries model/latency/token usage - never prompt or response content
```

See Governance's Audit section for wiring this into a durable, hash-
chained log alongside every other module's audit wrapper.

## Evaluation

A complete production evaluation surface for single- and multi-agent AI
systems - 48 metrics across 9 categories, organized one folder per
category under `agentic_ai.evaluation`:

| Category | Import | Measures |
|---|---|---|
| Agent | `agentic_ai.evaluation.agent` | Task Success/Correctness, Planning, Reasoning, Execution, Recovery, Autonomy, Loops, Lifecycle |
| LLM | `agentic_ai.evaluation.llm` | Response Correctness, Groundedness, Hallucination Rate, Instruction Following, Safety/Policy Violation, Latency/Tokens/Cost |
| Tools | `agentic_ai.evaluation.tools` | Selection/Argument Accuracy, Success/Failure Rate, Unnecessary Calls, Latency |
| Multi-Agent | `agentic_ai.evaluation.multi_agent` | Routing, Delegation, Handoff, Coordination, Duplicate Work |
| RAG | `agentic_ai.evaluation.rag` | Recall@K, Context Relevance, Groundedness, Citation Accuracy |
| Security | `agentic_ai.evaluation.security` | Prompt Injection, Unauthorized Execution, PII/Cross-Tenant Leakage, Authorization Violations |
| Platform | `agentic_ai.evaluation.platform` | Error Rate, Timeout Rate, Cost per Successful Task, SLA Compliance |
| Memory | `agentic_ai.evaluation.memory` | Retrieval Accuracy, Consistency |
| Drift | `agentic_ai.evaluation.drift` | Statistical (z-score) drift on Success/Correctness/Hallucination/Latency/Cost |

Every category is deterministic, LLM-judged, or a documented mix of
both - deterministic metrics need no LLM call at all (they read fields
you already populated); judged metrics reuse the same `LLMJudge` from
`agentic_ai.evaluation.core`, built on the gateway above, nothing else.

### Deterministic - no LLM call needed

```python
from agentic_ai.evaluation.agent import AgentTrace, compute_task_execution

traces = [
    AgentTrace(run_id="r1", task="find backend jobs", task_succeeded=True),
    AgentTrace(run_id="r2", task="find backend jobs", task_succeeded=False),
    AgentTrace(run_id="r3", task="find backend jobs", task_succeeded=True),
]
metrics = compute_task_execution(traces)
print(metrics.success_rate)  # 0.6666666666666666
```

### LLM-judged - needs a gateway, same one as above

```python
from agentic_ai.evaluation import LLMJudge
from agentic_ai.evaluation.llm import LLMCall, judge_response_correctness
from agentic_ai.gateway import LiteLLMGateway

judge = LLMJudge(LiteLLMGateway(), model="gpt-4o-mini")
call = LLMCall(call_id="c1", model="gpt-4o-mini", prompt="What is 2+2?", response="4")

result = judge_response_correctness(judge, call)
print(result.correct, result.score)
```

Every `judge_*()` function across every category takes an optional
`system_prompt` override - the built-in `DEFAULT_*` rubric is a real,
usable starting point, not the only valid one for every domain:

```python
from agentic_ai.evaluation.llm import judge_response_correctness

legal_rubric = "You are a strict legal-domain correctness judge. ..."
result = judge_response_correctness(judge, call, system_prompt=legal_rubric)
```

### Everything at once, persisted, compared over time

Agent Evaluation ties every deterministic + judged category together
into one report, storable and diffable:

```python
from agentic_ai.evaluation.agent import evaluate, JSONLEvaluationStore, compare

report = evaluate(traces, judge=judge)  # runs every computable category
store = JSONLEvaluationStore("eval_runs.jsonl")
store.save(report)

baseline = store.list_runs(limit=2)[1]
regressions = compare(baseline, report)  # direction-aware: knows failure_rate up is bad
```

For statistical drift across many runs over time (not just two points),
see `agentic_ai.evaluation.drift.compute_drift()` and its five named
wrappers (`compute_task_success_drift`, `compute_correctness_drift`,
`compute_hallucination_drift`, `compute_latency_drift`,
`compute_cost_drift`).

### Every category's own trace/call shape

`agent`, `llm`, `multi_agent`, `rag`, `security`, and `memory` each have
their own input model (`AgentTrace`, `LLMCall`, `MultiAgentTrace`,
`RAGQuery`, `AuthorizationCheck`/`TenantDataCheck`,
`MemoryRetrieval`) - populate the one your category needs from your own
agent's logging; nothing in this library runs an agent or a retriever
for you, it only evaluates the record you hand it.

## Memory & State

Ten memory types, each a small facade bound to a scope (a session id,
an agent id, a namespace) that knows its own purpose and picks sensible
defaults - backed by your choice of in-memory, file, SQLite, Redis,
Postgres (+pgvector), or Qdrant. No setup needed for local development;
pass a URL when you're ready for something durable.

| Facade | Import | Backs |
|---|---|---|
| `SessionMemory` | `agentic_ai.memory.SessionMemory` | Data scoped to one conversation/session |
| `AgentStateMemory` | `agentic_ai.memory.AgentStateMemory` | An agent's own operating state across turns |
| `ShortTermMemory` | `agentic_ai.memory.ShortTermMemory` | Recent context that outlives a single call |
| `WorkingMemory` | `agentic_ai.memory.WorkingMemory` | Scratch space for one in-flight task |
| `LongTermMemory` | `agentic_ai.memory.LongTermMemory` | Durable facts kept across sessions |
| `SemanticMemory` | `agentic_ai.memory.SemanticMemory` | Facts retrieved by meaning (vector search) |
| `EpisodicMemory` | `agentic_ai.memory.EpisodicMemory` | Past events/experiences, recallable by similarity |
| `ProceduralMemory` | `agentic_ai.memory.ProceduralMemory` | Versioned rules/workflows/operating procedures |
| `DocumentMemory` | `agentic_ai.memory.DocumentMemory` | Large source material (PDFs, contracts) + chunk search |
| `SharedMemory` | `agentic_ai.memory.SharedMemory` | A blackboard multiple agents read/write together |

### Zero-setup quickstart

```python
from agentic_ai.memory import SessionMemory

session = SessionMemory("session-42")  # defaults to an in-process InMemoryStore
session.set("last_intent", "book_flight", ttl_seconds=1800)
print(session.get("last_intent").value)  # "book_flight"
```

### Choosing a backend: just pass a URL

Every key/value facade accepts either an already-constructed store
(full control - `store=RedisStore(...)`, or a wrapper-composed one, see
below) or a plain shorthand - pick exactly one:

```python
from agentic_ai.memory import SessionMemory, LongTermMemory, AgentStateMemory

SessionMemory("session-42", redis_url="redis://localhost:6379/0")
LongTermMemory("user-123", postgres_url="postgresql://user:pass@localhost:5432/mydb")
AgentStateMemory("agent-7", sqlite_path="agent_state.db")
LongTermMemory("user-123", file_path="longterm.json")  # zero-setup but persisted to disk
```

`redis_url`/`postgres_url` need the matching install extra:

```bash
pip install 'enterprise-agentic-ai-framework[redis]'
pip install 'enterprise-agentic-ai-framework[postgres]'
pip install 'enterprise-agentic-ai-framework[qdrant]'
pip install 'enterprise-agentic-ai-framework[memory]'  # all three
```

Each facade connects and verifies immediately (a real ping / schema
init) - a bad URL fails fast in the constructor, not on some later,
unrelated call. This library is a client for all of these, never a
process manager: deploy Redis/Postgres/Qdrant yourself, same rule as
the LLM gateway above.

### Semantic, episodic, and document memory (vector-backed)

Embedding generation is always your job - these facades store and
search vectors, they never call an embedding model themselves:

```python
from agentic_ai.memory import SemanticMemory

memory = SemanticMemory("user-123", qdrant_url="http://localhost:6333", embedding_dim=1536)
memory.remember("pref-1", "prefers window seats", embedding=embed("prefers window seats"))
results = memory.recall(embed("seating preference"), top_k=3)
print(results[0].record.text, results[0].score)
```

`EpisodicMemory` and `DocumentMemory` combine a plain store (the log /
the raw document) with a vector index (similarity recall / chunk
search) - pass both explicitly, since a log store and a vector index
rarely share connection details:

```python
from agentic_ai.memory import EpisodicMemory
from agentic_ai.memory.stores.sqlite_store import SQLiteStore
from agentic_ai.memory.vector_stores.qdrant_store import QdrantVectorStore

episodes = EpisodicMemory(
    "agent-7",
    store=SQLiteStore("episodes.db"),
    vector_store=QdrantVectorStore(url="http://localhost:6333", embedding_dim=1536),
)
episodes.log_episode("ep-1", "deploy failed: dependency X unavailable", embedding=embed(...))
episodes.recall_similar(embed("deployment failure"), top_k=3)
```

`ProceduralMemory` is automatically versioned - every overwrite keeps
its prior value retrievable:

```python
from agentic_ai.memory import ProceduralMemory

procedures = ProceduralMemory("support-bot", postgres_url="postgresql://...")
procedures.set("refund_policy", {"max_days": 30, "requires_receipt": True})
procedures.get_history("refund_policy")  # every prior version, oldest first
```

### Governance, security, versioning, checkpointing, audit

Cross-cutting concerns are wrappers that compose onto any store, not
sixteen separate storage systems:

```python
from agentic_ai.memory.stores.redis_store import RedisStore
from agentic_ai.memory.wrappers import Actor, GovernedStore, RetentionPolicy, SecureStore
from agentic_ai.memory.core.models import MemoryType

base = RedisStore(url="redis://localhost:6379/0")

governed = GovernedStore(base, RetentionPolicy(
    default_ttl_seconds={MemoryType.SESSION: 1800},
    require_consent=True,
))
governed.set("session-42", "k", "v", memory_type=MemoryType.SESSION, consent=True)

secure = SecureStore(base, enforce_tenant_prefix=True)
actor = Actor(actor_id="u1", roles=["admin"], tenant_id="tenantA")
secure.set(actor, "tenantA:session-42", "k", "v")  # raises AccessDeniedError outside tenantA
```

`checkpoint()`/`restore()` snapshot and roll back a whole scope;
`VersionedStore` (what `ProceduralMemory` uses internally) keeps a
history on every write; `AuditedStore` emits an event to a sink you
provide for every operation, success or failure. All five live in
`agentic_ai.memory.wrappers` and take any `MemoryStore` - stack as many
as you need.

### One caveat: keys, not just types, need to be distinct

Every backend keys a record by `(scope, key)` only - `memory_type` is
stored on the record for filtering, not as part of the write key. Two
facades of different types that share both the same scope *and* the
same key on the same store will overwrite each other, same as two dict
writes to the same key would. In practice this is rarely an issue - one
scope with many distinct keys across several facade types is a normal,
safe pattern (`list()`/`clear()` on each facade only ever touch its own
`memory_type`).

## Context Engineering

Building the actual runtime context for one LLM call - the four
pillars (Write, Select, Compress, Isolate) plus one `core` layer that
ties them together: `assemble()`, the single function that takes
whatever candidate context you've gathered and produces a budget-
fitted, ordered, cache-boundary-marked result.

Everything in `select`/`compress`/`isolate` operates on plain
`ContextItem`s - a small model carrying content, its source
(provenance), a priority, an optional relevance score, and a trust
level - so every stage composes through the same shape instead of
each pillar inventing its own.

### Assembly - the core entry point

```python
from agentic_ai.context import assemble, ContextItem, ContextBudget

items = [
    ContextItem(id="sys", section="system", role="system", source="system_prompt",
                content="You are a booking assistant.", priority=1.0, cacheable=True),
    ContextItem(id="turn", section="conversation", source="conversation",
                content="Book me a flight to Denver.", priority=0.95),
]
budget = ContextBudget(max_tokens=4000, reserved_for_output=1000)
result = assemble(items, budget, section_order=["system", "memory", "tools", "conversation"])

print(result.total_tokens)
print([i.id for i in result.excluded])   # what got left out, and why - Context Observability
messages = result.to_messages()          # ready for agentic_ai.gateway.chat_completion(messages=...)
```

`assemble()` handles budget packing (drops lowest-priority items first,
per-section limits respected), ordering, conflict resolution (two items
sharing a `metadata["conflict_key"]` with different content - newest
wins by default), and cache-boundary marking (a `cacheable=True` item
only keeps that flag while it's part of an unbroken cacheable prefix,
since prompt caching only pays off on a shared, stable run of leading
content). Pass `allow_partial=False` to raise `ContextBudgetExceededError`
instead of silently dropping anything.

### Write - Scratchpad + Memory

Scratchpad is new: ephemeral, ordered notes for one run. "Memories" -
the part of Write meant to outlive the run - is `agentic_ai.memory`
itself; nothing here wraps it, there's nothing to add:

```python
from agentic_ai.context.write import Scratchpad

pad = Scratchpad("run-42", redis_url="redis://localhost:6379/0")  # same backend shorthand as agentic_ai.memory
pad.write("tried endpoint A, got a 404")
pad.write("trying endpoint B next")
pad.read_all()               # ordered ScratchpadEntry list
pad.to_context_items()       # ready for assemble()
```

### Select - relevance, prioritization, grounding, freshness, routing

```python
from agentic_ai.memory import SemanticMemory
from agentic_ai.context.select import from_vector_matches, prioritize, require_grounding, filter_stale

memory = SemanticMemory("user-123")
matches = memory.recall(embed("seating preference"), top_k=5)
items = from_vector_matches(matches, section="memory", source="memory:semantic")

grounded, _ = require_grounding(items)              # drops anything with no source
fresh, _ = filter_stale(grounded)                    # drops anything past its ttl_seconds
ranked = prioritize(fresh)                           # combines priority + relevance_score into one ranking
```

`select_by_relevance()` also ships a basic, dependency-free keyword
scorer for candidates that didn't come from a vector search; `route()`
fans a query out across several named sources (memory, a RAG index, a
tools catalog) and merges the results.

### Compress - trim, summarize, rolling summary

```python
from agentic_ai.context.compress import trim_to_budget, summarize_items, update_rolling_summary, RollingSummaryState

trimmed = trim_to_budget(candidate_items, max_tokens=2000)           # deterministic, no LLM call

from agentic_ai.gateway import LiteLLMGateway
gateway = LiteLLMGateway()
summary_item = summarize_items(gateway, "gpt-4o-mini", old_turns)     # LLM-based, for when trimming would cut load-bearing info

state = RollingSummaryState()
state = update_rolling_summary(gateway, "gpt-4o-mini", state, all_turns, keep_recent=10)
# keeps the last 10 turns verbatim + a running summary of everything older
```

### Isolate - trust boundaries, tenant scoping, sub-agent partitioning

```python
from agentic_ai.context.isolate import mark_trust_boundary, wrap_untrusted, enforce_tenant_scope, partition_context

marked = mark_trust_boundary(items, trusted_sources={"system_prompt", "memory:semantic"})
safe = [wrap_untrusted(i) for i in marked]   # delimits untrusted (e.g. tool/web) content so it can't pose as an instruction

allowed, _ = enforce_tenant_scope(items, tenant_id="tenantA")

# One assemble() per sub-agent, each with its own budget - one sub-agent's
# clutter never eats another's window:
results = partition_context(
    {"researcher": researcher_items, "writer": writer_items},
    {"researcher": ContextBudget(max_tokens=4000), "writer": ContextBudget(max_tokens=4000)},
)
```

`isolate.security` is structural, not a detection engine - it doesn't
classify content as an attack (that's the planned `guardrails`
module's job), it enforces what's already known: untrusted content
gets delimited, cross-tenant content gets filtered out.

### Observability - what context actually went to the model

```python
from agentic_ai.context import ContextTracer

tracer = ContextTracer("run-42", postgres_url="postgresql://...")  # same backend shorthand again
trace = tracer.record(result)   # result from assemble() above
tracer.list_traces()            # every trace recorded for this run, oldest first
```

## Secrets Management

A client for HashiCorp Vault - static (KV v2, versioned) secrets,
dynamic/leased credentials, and Transit encryption-as-a-service.
Deploy Vault yourself - this is a client, never a process manager,
same rule as every other real backend in this SDK.

```bash
pip install 'enterprise-agentic-ai-framework[vault]'
```

### Connecting

```python
from agentic_ai.secrets.vault import VaultClient

# Token auth - local dev, a CI job that already has one
vault = VaultClient(url="http://localhost:8200", token="s.xxxxx")

# AppRole - the standard machine/Workload Identity pattern
vault = VaultClient(url="https://vault.example.com:8200", role_id="...", secret_id="...")

# Kubernetes auth - reads the pod's own service account JWT, nothing
# else to configure when running in-cluster
vault = VaultClient(kubernetes_role="my-app")

# Or set nothing at all - VaultClient() reads VAULT_ADDR and, preferring
# a token if both are set, VAULT_TOKEN or VAULT_ROLE_ID+VAULT_SECRET_ID,
# the same environment variables the `vault` CLI itself uses
vault = VaultClient()
```

Connects and authenticates immediately - a bad token/role or an
unreachable Vault fails fast in the constructor, not on some later,
unrelated call. `mount_point` (default `"secret"`) and `namespace` let
one Vault serve several environments (`kv-dev`/`kv-prod` mounts, or
Vault Enterprise namespaces) - Environment Isolation and Cross-
environment management, without a separate abstraction.

### Static secrets (KV v2) - Secret Versioning built in

```python
vault.set_secret("db/prod", {"username": "app", "password": "s3cr3t"})
secret = vault.get_secret("db/prod")
print(secret)          # Secret(path='db/prod', version=1, keys=['password', 'username']) - never the values
print(secret.data)     # {"username": "app", "password": "s3cr3t"} - only here, on purpose

vault.set_secret("db/prod", {"username": "app", "password": "rotated"})
old = vault.get_secret("db/prod", version=1)      # any prior version, still readable
vault.rollback_secret("db/prod", to_version=1)     # writes it back as a NEW version - reversible, not a silent revert

vault.delete_secret("db/prod")                     # soft delete, recoverable
vault.destroy_secret_versions("db/prod", [1, 2])   # permanent - specific versions
vault.purge_secret("db/prod")                      # permanent - everything
```

`Secret.__repr__`/`__str__` never include the actual values, only key
names - a safety net for the common `print(secret)`/`log.info("%s", secret)`
mistake; `secret.data` still gives you the real values, since that's
the entire point of fetching a secret.

### Dynamic/leased secrets

```python
creds = vault.read_dynamic_secret("database/creds/readonly")  # any dynamic secrets engine - generic, not engine-specific
print(creds.lease_id, creds.lease_duration, creds.renewable)

vault.renew_lease(creds.lease_id, increment=3600)  # Secret Rotation via renewal, TTL/Expiration
vault.revoke_lease(creds.lease_id)                 # done early - revoke rather than wait out the TTL
```

### Transit - encryption as a service

The key material never leaves Vault; this process only ever sees
ciphertext:

```python
vault.create_transit_key("app-data")
ciphertext = vault.encrypt("app-data", "a value worth encrypting")   # -> "vault:v1:..."
plaintext = vault.decrypt_text("app-data", ciphertext)                # -> "a value worth encrypting"
```

Sign/verify uses the same engine for content provenance instead of
secrecy - proving a system prompt, a tool definition, or any other
content is unmodified and actually came from whoever holds the key
(Prompt Identity), without that key ever leaving Vault. Needs a
signing-capable key type (`key_type="ed25519"` or similar - the
default AES key `create_transit_key()` makes for encrypt/decrypt can't
sign):

```python
vault.create_transit_key("prompt-signing-key", key_type="ed25519")
signature = vault.sign("prompt-signing-key", system_prompt_text)
vault.verify("prompt-signing-key", system_prompt_text, signature)  # True/False, never raises for a bad signature
```

### Caching and audit logging

Cross-cutting concerns are wrappers, same pattern as
`agentic_ai.memory`'s - they work over `VaultClient` or any future
backend implementing the same small `SecretsProvider` protocol:

```python
from agentic_ai.secrets import CachedSecretsProvider, AuditedSecretsProvider

cached = CachedSecretsProvider(vault, ttl_seconds=60)   # avoid hitting Vault on every call; never persisted to disk
cached.get_secret("db/prod")

def sink(event):
    logger.info("secret access", extra=event.model_dump())

audited = AuditedSecretsProvider(vault, sink=sink, actor_id="checkout-service")
audited.get_secret("db/prod")  # event carries path/action/version/outcome - never the secret's value
```

This client-side audit trail is distinct from Vault's own server-side
audit log (enable that too - `vault audit enable file file_path=...` -
it captures every raw API call Vault receives, independent of this
SDK); this wrapper lets *your application* route access events into
its own observability pipeline. RBAC/Policies are entirely Vault-side:
whatever policy your token/AppRole/Kubernetes role carries is what
this client can and can't do - a disallowed action surfaces as
`SecretAuthError`, not a silent no-op.

## Guardrails & Content Safety

Input/output validation, prompt injection and jailbreak defense,
content safety, PII/DLP detection and redaction, topic restrictions,
tool allow/deny-listing, rate limiting, and human-in-the-loop approval
- all composed through one `GuardrailPipeline`. There's no separate
"Input Guardrails" vs. "Output Guardrails" class: the same pipeline
runs before the LLM call (input) or after it (output) - which one
you're doing is just where you call `run()`.

Scope note: this module owns detection and enforcement at the content
layer. Full RBAC/tenant authorization lives in `agentic_ai.memory`'s
`SecureStore` and the future `governance` module; untrusted-content
isolation and grounding checks are `agentic_ai.context.isolate` and
`agentic_ai.context.select.grounding`; encryption and workload identity
are `agentic_ai.secrets.vault`. Guardrails composes with those, it
doesn't re-implement them.

### The pipeline

A rule is any `Callable[[str, dict | None], list[Finding]]` - every
detector below already has that shape (curry in a judge with
`functools.partial` where one's needed), and so does any custom
callable you write:

```python
from agentic_ai.guardrails import GuardrailPipeline
from agentic_ai.guardrails.detectors import detect_prompt_injection_heuristic, detect_sensitive_data

input_guard = GuardrailPipeline([detect_prompt_injection_heuristic, detect_sensitive_data], name="input")

result = input_guard.run("my email is jane@example.com, can you reset my password?")
print(result.passed)              # True - PII is sanitized, not blocked
print(result.action)              # GuardrailAction.SANITIZE
print(result.sanitized_content)   # "my email is [REDACTED_EMAIL], can you reset my password?"
```

`GuardrailResult.passed` is True for ALLOW/SANITIZE (content can
proceed, possibly modified) and False for ESCALATE/BLOCK. Pass
`raise_on_block=True` to get a `GuardrailBlocked` exception instead of
checking `result.passed` yourself.

### Detectors

| Detector | Import | Mechanism |
|---|---|---|
| Prompt injection | `detect_prompt_injection_heuristic`, `judge_prompt_injection` | phrase pre-filter + LLM judge |
| Jailbreak | `detect_jailbreak_heuristic`, `judge_jailbreak` | phrase pre-filter + LLM judge |
| Content safety | `judge_content_safety` | LLM judge (harassment/hate, violence, sexual, self-harm, dangerous/illegal) |
| Topic restriction | `judge_topic_restriction` | LLM judge |
| PII | `detect_pii` | regex (email, phone, SSN, credit card) |
| Confidential data | `detect_confidential_data` | regex (AWS keys, GitHub/Slack tokens, generic API keys) |

The phrase pre-filters are zero-cost and catch the obvious cases; they
are not the real detector for anything subtler - pair them with the
LLM-judged versions, built on `agentic_ai.evaluation.LLMJudge` (reused,
not rebuilt - the same JSON-mode judge engine every measurement judge
in this SDK already uses):

```python
from functools import partial
from agentic_ai.evaluation import LLMJudge
from agentic_ai.gateway import LiteLLMGateway
from agentic_ai.guardrails.detectors import judge_prompt_injection, judge_content_safety

judge = LLMJudge(LiteLLMGateway(), model="gpt-4o-mini")
pipeline = GuardrailPipeline([
    detect_prompt_injection_heuristic,
    partial(judge_prompt_injection, judge),
    partial(judge_content_safety, judge),
])
```

Every judge function takes an optional `system_prompt` override, same
pattern as every judge in `agentic_ai.evaluation`. `Finding.detail`
never contains the matched text itself, only a safe-to-log summary and
a `span` into your own original content - a Finding can be handed to
an audit sink without a second thought.

### Tool / Action guardrails

Deliberately a lightweight allow/deny-list, not a full RBAC engine:

```python
from agentic_ai.guardrails.tools import ToolPolicy, check_tool_call

policy = ToolPolicy(allowed_tools={"search", "reset_password"}, denied_tools={"delete_account"})
result = check_tool_call(policy, "reset_password", {"user_id": "u1"}, actor="session-1")
```

### Rate limiting and human-in-the-loop

Both reuse `agentic_ai.memory` for their state - same backend
shorthand (`redis_url`, `postgres_url`, ...) as everywhere else in this
SDK, so a limit or an approval queue is durable and shared across
processes when you need it to be:

```python
from agentic_ai.guardrails.abuse import RateLimiter
from agentic_ai.guardrails.human import ApprovalQueue

limiter = RateLimiter(limit=100, window_seconds=60, redis_url="redis://localhost:6379/0")
if not limiter.check("user-42").allowed:
    ...

queue = ApprovalQueue(postgres_url="postgresql://...")
request = queue.request_approval("wire transfer of $50,000", risk_level="critical", requested_by="agent-7")
# a separate reviewer process/UI calls queue.approve(request.request_id, resolved_by="alice")
```

### Auditability

```python
from agentic_ai.guardrails.wrappers import AuditedGuardrailPipeline

audited = AuditedGuardrailPipeline(input_guard, sink=lambda e: logger.info("guardrail", extra=e.model_dump()), actor_id="session-1")
audited.run(user_message)  # event carries rule/category/severity/action - never the raw content or a sensitive span
```

## Authentication

A client for Keycloak's authentication mechanics - the "how do you
prove it" layer: password/SSO/OIDC/OAuth2 grants, MFA, mTLS client-
certificate auth, token issuance/validation/refresh/revocation. Not
authorization: it proves who someone (or some service) is and hands
back a token plus raw claims data. Interpreting those claims into a
rich identity is `agentic_ai.identity`'s job (below), built from this
client's output - this module has no dependency on `identity` at all.
Enforcing what an identity is allowed to do is a further step still,
the future `governance` module's job, same boundary guardrails' tool
allow-listing already draws.

```bash
pip install 'enterprise-agentic-ai-framework[authentication]'
```

### Connecting and authenticating

```python
from agentic_ai.authentication import KeycloakClient

client = KeycloakClient(
    server_url="http://localhost:8080", realm_name="my-realm",
    client_id="my-service", client_secret="...",
)

# Password grant - a user's own credentials (add totp=123456 for MFA realms)
tokens = client.authenticate(username="alice", password="...")

# client_credentials grant - Workload Identity for a service, no user involved
tokens = client.authenticate_service_account()

print(tokens.access_token, tokens.expires_at, tokens.is_expired)  # tokens.__repr__ never shows the actual strings
```

Connects and confirms the realm exists immediately - a bad URL or
unknown realm fails fast in the constructor, not on some later,
unrelated call. Pass `client_cert`/`client_key` (PEM file paths)
instead of/alongside `client_secret` for Certificate Authentication
(mTLS) to Keycloak itself.

### Validating a token

```python
raw_claims = client.decode_token(tokens.access_token)   # local, offline - verifies signature against Keycloak's JWKS; returns a plain dict
status = client.introspect(tokens.access_token)          # asks Keycloak directly - catches server-side revocation decode_token() can't see
print(status.active)

raw_user_info = client.get_user_info(tokens.access_token) # the userinfo endpoint; also a plain dict
```

`decode_token()`/`get_user_info()` deliberately return raw dicts, not
identity objects - see Identity below for the typed models built from
this output.

### Refresh, revoke, logout, and the authorization_code flow

```python
refreshed = client.refresh(tokens.refresh_token)
client.revoke_token(tokens.access_token)  # RFC 7009 - revokes a specific token, distinct from logout()
client.logout(tokens.refresh_token)        # ends the session via its refresh token

url = client.build_authorization_url("https://app.example.com/callback", state="xyz")
# ... redirect the user's browser to `url` - Keycloak's own hosted login
# page handles SSO/WebAuthn/FIDO2 transparently here, nothing extra to
# implement - they come back with a `code` ...
tokens = client.exchange_code_for_token(code, "https://app.example.com/callback")
```

### Agent / Tools Identity - Token Exchange

OAuth2 Token Exchange (RFC 8693) - the standards-based way to give an
agent or a tool call its own distinct, scoped, attributable token
derived from the original caller's, instead of every agent/tool call
sharing one client's identity. Requires Keycloak's Standard Token
Exchange feature enabled for the client:

```python
agent_tokens = client.exchange_token(user_tokens.access_token, audience="downstream-service")
```

### Auto-refresh

The tedious part of every OAuth client, done once:

```python
from agentic_ai.authentication import TokenManager

manager = TokenManager(client, grant_type="client_credentials")
header = {"Authorization": f"Bearer {manager.get_access_token()}"}
# subsequent calls authenticate on first use, refresh near expiry, and
# fall back to a full re-authentication if the refresh token itself
# has expired - manager.invalidate() forces a fresh one, e.g. after a
# downstream 401
```

### Auditability

```python
from agentic_ai.authentication.wrappers import AuditedKeycloakClient

audited = AuditedKeycloakClient(client, sink=lambda e: logger.info("auth", extra=e.model_dump()))
audited.authenticate(username="alice", password="...")  # event carries grant_type/username/outcome - never the token or password
```

## Identity

The claims/data layer, built entirely from plain dicts - typically
`agentic_ai.authentication`'s `decode_token()`/`get_user_info()`
output, or any other OIDC-compatible source. No dependency on
`authentication`, or on `python-keycloak` being installed at all.

```python
from agentic_ai.authentication import KeycloakClient
from agentic_ai.identity import TokenClaims, UserInfo

auth = KeycloakClient(server_url="...", realm_name="my-realm", client_id="my-service", client_secret="...")
tokens = auth.authenticate(username="alice", password="...")

claims = TokenClaims.from_claims(auth.decode_token(tokens.access_token))
print(claims.subject, claims.username, claims.realm_roles)
print(claims.has_realm_role("admin"), claims.has_client_role("my-service", "editor"))

info = UserInfo.from_response(auth.get_user_info(tokens.access_token))
```

### Tenant Identity Isolation

Keycloak has no built-in "tenant" claim - if your deployment adds one
via a custom protocol mapper, `TokenClaims` picks it up automatically
(`claims.tenant_id`), and `require_tenant()` enforces it, same pattern
as `memory.wrappers.SecureStore`'s `enforce_tenant_prefix` and
`guardrails.isolate.enforce_tenant_scope`:

```python
from agentic_ai.identity import require_tenant, TenantMismatchError

require_tenant(claims, "tenant-A")  # raises TenantMismatchError on any other tenant
# a differently-named claim: require_tenant(claims, "org-42", claim_name="org_id")
```

If your deployment instead models tenancy as one realm per tenant,
isolation is already realm boundaries - construct a separate
`KeycloakClient` per tenant and this helper isn't the right tool for
that shape.

## Governance

`agentic_ai.governance` is the umbrella for this SDK's policy/
compliance components, each its own subpackage rather than flattened
into one namespace. `authorization` is implemented today (below);
`compliance`, `audit`, `data_governance`, `finops`, `responsible_ai`,
and `security` are scaffolded, not yet usable.

Deliberately not here: `agentic_ai.identity` and
`agentic_ai.authentication` (IAM - who someone is and proving it) stay
top-level siblings of this package, not nested inside it -
`governance.authorization` *consumes* identity as input
(`user.roles`), so nesting identity inside governance would put the
dependency direction backwards.

### Authorization

A client for Open Policy Agent's REST API - the Policy Decision Point
(PDP) `authorize()`/`evaluate()` query. RBAC, ABAC, fine-grained,
resource-, action-, agent-, tool-, and tenant-level authorization are
all just different shapes of one generic policy query, not separate
code paths - that's OPA's own design, not something this client
reinvents. Nothing is hardcoded: the OPA URL, the policy path, and
every piece of the authorization input are parameters you supply.

Built directly on `httpx` (already a base dependency, same as
`agentic_ai.gateway`) - no new extra needed.

```python
from agentic_ai.governance.authorization import OPAClient

opa = OPAClient(url="http://localhost:8181")  # or host="...", port=8181 - never a fixed address
```

Connects and verifies immediately (a real `/health` check) - an
unreachable OPA fails fast in the constructor.

```python
decision = opa.authorize(
    user={"roles": ["job-seeker"]},
    action="invoke",
    resource={"type": "agent", "name": "job_search"},
    context={"risk_score": 0, "business_hours": True},
    policy_path="authz/allow",  # always overridable - no assumption about your package layout
)
print(decision.allowed, decision.defined)  # defined=False means "no such policy", not "policy said no"

opa.authorize(..., raise_on_deny=True)  # raises AuthorizationDenied instead of checking .allowed yourself

# Any result shape - not just plain allow/deny:
decision = opa.evaluate("authz", {"user": ..., "action": ..., "resource": ...})
print(decision.result)  # e.g. {"allow": true, "risk_threshold": 2}
```

### Dynamic context - never a static risk_score/business_hours

```python
from agentic_ai.governance.authorization.context import build_context

context = build_context(
    risk_score=0,                       # a plain int, or a zero-arg callable computed fresh per call
    business_hours_start=9, business_hours_end=17,  # default 0-24 (wide open) - narrow explicitly, never assumed
    tenant_id="tenant-A",                # any **extra merges straight into the context dict
)
```

Wire a real risk score from `agentic_ai.guardrails`' own audit trail
instead of a fixed number:

```python
from functools import partial
from agentic_ai.governance.authorization.risk import count_recent_violations

risk_score = partial(count_recent_violations, recent_events, actor_id=user_id, window_seconds=3600)
context = build_context(risk_score=risk_score)
```

### Policy management

```python
opa.upload_policy("authz", rego_source)
opa.list_policies()
opa.get_policy("authz")
opa.delete_policy("authz")
```

### Auditability

```python
from agentic_ai.governance.authorization.wrappers import AuditedOPAClient

audited = AuditedOPAClient(opa, sink=lambda e: logger.info("authz", extra=e.model_dump()))
audited.authorize(user=..., action=..., resource=...)  # event carries subject/action/resource/allowed - context excluded unless include_context=True
```

### What's already covered elsewhere, not rebuilt here

RBAC (`agentic_ai.identity.TokenClaims`), Delegated Authorization
(`agentic_ai.authentication.KeycloakClient.exchange_token()`), Tenant
Authorization (`agentic_ai.identity.require_tenant()`), Tool
Authorization (`agentic_ai.guardrails.tools.ToolPolicy`), and Human
Approval (`agentic_ai.guardrails.human.ApprovalQueue`) already exist -
this module composes with them. Transaction Limits and Data
Authorization need no special code - they're just more fields in
`resource`/`context` for your policy to check. Least Privilege is a
principle for how you write your Rego policies, not something a client
enforces.

### Audit

Every audit wrapper across this SDK (`memory`, `secrets`, `guardrails`,
`authentication`, `governance.authorization`, `gateway`) hands you an
event via a `sink` callback and deliberately leaves storage to you -
`AuditLog` is a real, durable place to put them: hash-chained (each
record links to the one before it, so tampering, deletion, or
reordering is detectable), backed by `agentic_ai.memory` (same "just
pass a url" pattern as everywhere else - no new dependency).

```python
from agentic_ai.governance.audit import AuditLog

log = AuditLog("platform", redis_url="redis://localhost:6379/0")  # or in-memory/file/sqlite/postgres - same shorthand as everywhere else
```

Wire any existing wrapper's `sink` straight into it with the matching adapter:

```python
from agentic_ai.governance.audit.adapters import guardrail_sink, llm_sink
from agentic_ai.guardrails.wrappers import AuditedGuardrailPipeline
from agentic_ai.gateway.wrappers import AuditedLiteLLMGateway

audited_pipeline = AuditedGuardrailPipeline(input_guard, sink=guardrail_sink(log))
audited_gateway = AuditedLiteLLMGateway(gateway, sink=llm_sink(log))  # LLM Audit - model, latency, token usage, never prompt/response content
```

`memory_sink`, `secret_sink`, `guardrail_sink`, `authentication_sink`,
`authorization_sink`, and `llm_sink` cover every existing audit
wrapper; `log_approval_decision(log, resolved_request)` is a
convenience for `agentic_ai.guardrails.human.ApprovalQueue` (it
returns the resolved decision directly rather than streaming through a
callback). Agent, Multi-Agent, Tool, Data/RAG, and Business-
Transaction audit have no dedicated adapter - this SDK doesn't itself
execute those, so call `log.append(source_module=..., action=...,
outcome=..., actor_id=..., details=...)` directly.

```python
records = log.list_records(actor_id="user-1", since=one_hour_ago)
log.verify_chain()  # raises AuditIntegrityError if anything was tampered with, deleted, or reordered
```

### Security

One submodule per domain - `agentic_ai.governance.security.<domain>`,
each its own import (several would collide on names flattened into
one namespace). `llm` and `tool` are thin, documented re-exports of
`agentic_ai.guardrails` (that's where the real detection/allow-listing
logic lives); every other submodule is new. Not here at all -
genuinely not something a Python client library can provide: Network,
Infrastructure, Cloud, and OS-level Runtime Security, Supply Chain
Security, and Vulnerability Management (SAST/DAST/scanning/
pentesting).

```python
# Agent Security - capability restrictions + autonomous-step limits (new)
from agentic_ai.governance.security.agent import AgentSecurityPolicy, check_agent_action

policy = AgentSecurityPolicy(allowed_actions={"search", "book"}, max_autonomous_steps=10)
result = check_agent_action(policy, "search", step_count=3, agent_id="agent-7")

# Multi-Agent Security - trust boundaries between agents (new)
from agentic_ai.governance.security.multi_agent import AgentTrustPolicy, require_trusted_agent

trust = AgentTrustPolicy(trusted_agent_ids={"orchestrator", "researcher"})
require_trusted_agent(trust, sender_agent_id)  # raises UntrustedAgentError otherwise

# Memory Security - Memory Poisoning Protection: screen content before it's written (new)
from agentic_ai.governance.security.memory import ScreenedStore

screened = ScreenedStore(redis_store, input_guard)  # any MemoryStore + any GuardrailPipeline
screened.set("agent-7", "learned_fact", tool_output_text)  # raises MemoryWriteBlocked if the pipeline blocks it

# RAG Security - screen retrieved documents the same way input/output is screened (new)
from agentic_ai.governance.security.rag import screen_retrieved_document

result = screen_retrieved_document(input_guard, retrieved_document_text, source="knowledge_base/doc1")

# Privacy Security - PII redaction + consent + retention as one store (new)
from agentic_ai.governance.security.privacy import PrivacyStore
from agentic_ai.memory.wrappers.governance import RetentionPolicy

privacy_store = PrivacyStore(redis_store, RetentionPolicy(require_consent=True))
privacy_store.set("user-1", "note", "contact jane@example.com", consent=True)  # -> "[REDACTED_EMAIL]"

# Incident Response - kill switch (new)
from agentic_ai.governance.security.kill_switch import KillSwitch

kill_switch = KillSwitch("platform", redis_url="redis://localhost:6379/0")  # shared across every process
kill_switch.check()  # raises KillSwitchEngaged if someone has engaged it - call this at the start of any haltable operation
kill_switch.engage(reason="prompt injection wave detected", actor_id="oncall-1")

# Security Monitoring - SIEM export in CEF (new)
from agentic_ai.governance.security.siem import export_cef

cef_lines = export_cef(log.list_records())  # ready to ship to ArcSight/Splunk/QRadar

# Red Teaming - self-test your own guardrails against known adversarial payloads (new)
from agentic_ai.governance.security.redteam import run_red_team, pass_rate

results = run_red_team(input_guard)  # DEFAULT_PAYLOADS, or pass your own `cases`
print(pass_rate(results))
```

### Data Governance & Security

One submodule per domain - `agentic_ai.governance.data.<domain>`.
`classification`, `access`, `ai_access`, `lineage`, `quality`,
`lifecycle`, and `rag` are new code. `privacy`, `dlp`, `encryption`,
`isolation`, and `retention` are thin, documented re-exports of where
the real implementation already lives (`agentic_ai.guardrails.detectors.pii`,
`agentic_ai.secrets.vault`, `agentic_ai.identity`,
`agentic_ai.memory.wrappers`) - not duplicated, just made discoverable
under the Data Governance name. Not here at all - genuinely not
something a Python client library can provide: physical data residency
enforcement, database-level column encryption/masking, and full
ETL-grade data-quality profiling.

```python
# Data Classification - the sensitivity tier everything else keys off of (new)
from agentic_ai.governance.data.classification import DataClassification, ClassificationPolicy, HandlingRequirements

policy = ClassificationPolicy(requirements={
    DataClassification.RESTRICTED: HandlingRequirements(
        minimum_role="admin", requires_encryption=True, default_retention_seconds=3600,
    ),
})

# Data Access Governance - does this actor's role meet the classification's minimum (new)
from agentic_ai.governance.data.access import check_data_access

result = check_data_access(policy, DataClassification.RESTRICTED, {"employee"}, actor_id="user-1")
result.passed  # False - "employee" doesn't meet the "admin" minimum_role for RESTRICTED data

# AI Data Access Control - the same check, named for an agent/model caller (new)
from agentic_ai.governance.data.ai_access import check_ai_data_access

check_ai_data_access(policy, DataClassification.RESTRICTED, {"admin"}, agent_id="agent-1").passed  # True

# Data Privacy & PII / Data Loss Prevention - both bridge to agentic_ai.guardrails.detectors.pii
from agentic_ai.governance.data.privacy import detect_sensitive_data, redact

findings = detect_sensitive_data("contact jane@example.com")
redact("contact jane@example.com", findings)  # -> "contact [REDACTED_EMAIL]"

# Data Encryption - thin pass-through to a VaultClient-shaped object, no hvac import forced on you
from agentic_ai.governance.data.encryption import encrypt_data, decrypt_data

ciphertext = encrypt_data(vault_client, "app-key", "sensitive-value")
decrypt_data(vault_client, "app-key", ciphertext)  # -> "sensitive-value"

# Data Isolation - structural tenant isolation, bridges to agentic_ai.memory.wrappers.security.SecureStore
from agentic_ai.governance.data.isolation import Actor, SecureStore
from agentic_ai.memory.stores.in_memory import InMemoryStore

secure_store = SecureStore(InMemoryStore(), enforce_tenant_prefix=True)
secure_store.set(Actor(actor_id="u1", tenant_id="tenant-a"), "tenant-a:docs", "k1", "v1")  # a tenant literally cannot reach another tenant's scope

# Data Retention & Deletion - bridges to agentic_ai.memory.wrappers.governance.GovernedStore
from agentic_ai.governance.data.retention import GovernedStore, RetentionPolicy
from agentic_ai.memory.core.models import MemoryType

governed_store = GovernedStore(InMemoryStore(), RetentionPolicy(default_ttl_seconds={MemoryType.SESSION: 3600}))
governed_store.set("scope1", "k1", "v1", memory_type=MemoryType.SESSION)  # expires in 3600s by default

# Data Lineage - what data descended from what, distinct from governance.audit's who-did-what (new)
from agentic_ai.governance.data.lineage import DataLineage

lineage = DataLineage("platform", redis_url="redis://localhost:6379/0")  # or in-memory/file/sqlite/postgres, same shorthand as everywhere else
lineage.record("raw-doc-1", transformation="ingest")
lineage.record("summary-1", source_ids=["raw-doc-1"], transformation="summarize")
lineage.trace_ancestry("summary-1")  # -> [summary-1, raw-doc-1], closest ancestor first

# Data Quality - a lightweight rule-runner, not a full ETL/data-quality platform (new)
from agentic_ai.governance.data.quality import required_fields_rule, run_quality_checks

rule = required_fields_rule("has_name_email", ["name", "email"])
run_quality_checks({"name": "Jane"}, [rule])  # -> [DataQualityResult(rule_name="has_name_email", passed=False, ...)]

# Data Lifecycle Management - derives retention from classification instead of configuring both separately (new)
from agentic_ai.governance.data.lifecycle import retention_policy_for

retention_policy_for(policy, {DataClassification.RESTRICTED: MemoryType.LONG_TERM})  # -> RetentionPolicy(default_ttl_seconds={LONG_TERM: 3600})

# RAG Data Governance - a retrieved document must pass content screening AND classification-based access (new)
from agentic_ai.governance.data.rag import govern_rag_retrieval

result = govern_rag_retrieval(
    input_guard, retrieved_document_text,
    classification_policy=policy, document_classification=DataClassification.INTERNAL, actor_roles={"employee"},
)
```

### Compliance

A general-purpose evidence register - not AI-specific, since
compliance requirements (SOC 2, ISO 27001, GDPR, the EU AI Act, ...)
apply across a whole platform. `agentic_ai.governance.responsibleai`
bridges this in under the "AI Governance & Compliance" name rather
than duplicating a second register there.

```python
from agentic_ai.governance.compliance import ComplianceRegister, ComplianceStatus

register = ComplianceRegister("platform", redis_url="redis://localhost:6379/0")  # or in-memory/file/sqlite/postgres, same shorthand as everywhere else
register.record("SOC 2 CC6.1", framework="SOC 2", status=ComplianceStatus.COMPLIANT, evidence="access review Q3-2026")

register.latest_status("SOC 2 CC6.1")  # the most recent record - every record() call is kept as history, not overwritten
register.require_compliant("SOC 2 CC6.1")  # raises ComplianceRequirementMissing unless the latest status is COMPLIANT
```

### Responsible AI

One submodule per capability - `agentic_ai.governance.responsibleai.<domain>`.
`risk`, `explainability`, and `fairness` are new code; `oversight`
bridges human approval from `agentic_ai.guardrails.human` (accountability
tracking is new); `safety` bridges the kill switch/timeout from
`agentic_ai.governance.security` and judge-reliability checks from
`agentic_ai.evaluation` (the operating-envelope check is new);
`compliance` bridges `agentic_ai.governance.compliance` and
`agentic_ai.governance.audit`. Not here at all - genuinely not
something a Python client library can provide: certifying an AI system
as fair, safe, or compliant in fact. This module gives you the
structural tools; the judgment calls (what tier is this use case, what
bias threshold is acceptable, who is actually accountable) stay yours.

```python
# 1. AI Risk Management - classify a use case's risk tier, gate on it (new)
from agentic_ai.governance.responsibleai.risk import AIRiskTier, RiskRegister, require_permitted_tier

risk_register = RiskRegister("platform")
assessment = risk_register.record(
    "resume-screening", AIRiskTier.HIGH, rationale="automated hiring decisions",
    mitigations=["human review required"], owner="hr-ai-team",
)
require_permitted_tier(assessment, AIRiskTier.LIMITED)  # raises RiskThresholdExceeded - HIGH exceeds LIMITED

# 2. Human Oversight & Accountability - approval is agentic_ai.guardrails.human's ApprovalQueue; ownership is new
from agentic_ai.governance.responsibleai.oversight import AccountabilityRegister, ApprovalQueue

accountability = AccountabilityRegister("platform")
accountability.assign_owner("resume-screening-model", "alice@corp.com", role="model owner")
accountability.require_owner("resume-screening-model")  # raises AccountabilityRequired if nobody's on record

approvals = ApprovalQueue(scope="platform_approvals")
request = approvals.request_approval("deploy resume-screening-model v2", requested_by="alice@corp.com")
approvals.approve(request.request_id, resolved_by="bob@corp.com")

# 3. Transparency & Explainability - system-level documentation + per-decision evidence (new)
from agentic_ai.governance.responsibleai.explainability import SystemCard, explain_decision, judge_explanation_quality

card = SystemCard(
    system_name="resume-screener", version="2.0", intended_use="rank candidate resumes for recruiter review",
    limitations=["not validated outside US English resumes"], owner="hr-ai-team",
)
explanation = explain_decision(
    "decision-123", "Candidate ranked #3 due to matching 8/10 required skills.",
    evidence=["skill_match=0.8", "years_experience=5"], confidence=0.72, model="gpt-4o-mini",
)
judge_explanation_quality(judge, explanation)  # LLMJudge - flags generic/boilerplate explanations

# 4. Fairness & Bias - an LLM-judged check for one response, a deterministic metric for outcome data (new)
from agentic_ai.governance.responsibleai.fairness import judge_fairness, outcome_parity

judge_fairness(judge, model_output_text)  # is_biased, protected_attributes, confidence, reasoning
outcome_parity({"group_a": [True, True, True, False, False], "group_b": [True, False, False, False, False]})
# -> impact_ratio=0.33, passes_four_fifths_rule=False (the US EEOC's disparate-impact threshold)

# 5. AI Safety & Reliability - a structural operating-envelope check (new) + kill switch/timeout (bridged)
from agentic_ai.governance.responsibleai.safety import SafetyPolicy, check_safety_bounds, run_with_timeout, KillSwitch

policy = SafetyPolicy(max_retries=3, max_output_length=2000)
check_safety_bounds(policy, model_output_text, attempt=1)  # [] if within bounds, else violation messages

kill_switch = KillSwitch("platform", redis_url="redis://localhost:6379/0")
kill_switch.check()  # raises KillSwitchEngaged if someone has halted the platform

# 6. AI Governance & Compliance - policy evidence + audit trail, both bridged
from agentic_ai.governance.responsibleai.compliance import ComplianceRegister, ComplianceStatus, AuditLog

compliance_register = ComplianceRegister("platform_ai")
compliance_register.record("EU AI Act Art. 9", framework="EU AI Act", status=ComplianceStatus.COMPLIANT, evidence="risk management system documented")

audit_log = AuditLog("platform_ai_audit")
audit_log.append(source_module="responsibleai", action="risk_assessed", outcome="success", actor_id="alice@corp.com", details={"use_case": "resume-screening", "tier": "high"})
```

### FinOps

One submodule per capability - `agentic_ai.governance.finops.<domain>`.
All new code except where it deliberately fills a gap another module's
docstring already named (LLMJudge's cost computation). `CostLedger` and
`BudgetRegister` persist via `agentic_ai.memory`, same "just pass a
url" pattern as every other real backend in this SDK - point them at
**`postgres_url`** for durable, queryable, BI-joinable cost data; that's
the backend this module is built to be used with in production, not
just one option among several. Not here at all - genuinely not
something a Python client library can do to actual cloud/GPU/database
infrastructure: real resource optimization requires your cloud
provider's own cost-optimization API, or a human. `recommend_rightsizing()`
below is the honest, generic arithmetic that stands in for it.

```python
# 1. Cost Monitoring & Visibility - a durable, queryable ledger across every resource type
from agentic_ai.governance.finops.ledger import CostLedger
from agentic_ai.governance.finops.models import ResourceType

ledger = CostLedger("platform", postgres_url="postgresql://user:pass@localhost:5432/finops")  # or in-memory/redis/sqlite/file for smaller deployments
ledger.record_cost(ResourceType.LLM, 1.25, quantity=5000, unit="tokens", dimensions={"team": "growth", "model": "gpt-4o-mini", "tenant_id": "acme"})
ledger.record_cost(ResourceType.GPU, 4.00, quantity=2, unit="gpu_hours", dimensions={"team": "ml-platform"})
ledger.total_cost()  # -> 5.25

# 2. Cost Allocation & Attribution - group the same ledger by whatever dimension matters
from agentic_ai.governance.finops.attribution import attribute_costs

attribute_costs(ledger, "team")  # -> {"growth": 1.25, "ml-platform": 4.00}

# 3. Budget & Cost Controls - a spending limit scoped to a dimension filter, checked against the ledger
from agentic_ai.governance.finops.budget import BudgetRegister

budgets = BudgetRegister("platform", postgres_url="postgresql://user:pass@localhost:5432/finops")
growth_budget = budgets.create("growth-team-monthly", 100.0, scope={"team": "growth"}, alert_threshold_pct=0.8)
budgets.check(growth_budget, ledger, raise_on_exceeded=True)  # raises BudgetExceeded once spend crosses limit_usd

# 4. LLM / Model Cost Optimization - the per-model pricing table agentic_ai.evaluation.LLMJudge's own docstring defers to this module
from agentic_ai.governance.finops.pricing import ModelPricing, PricingTable, compute_llm_cost, estimate_cache_savings

pricing = PricingTable(rates={"gpt-4o-mini": ModelPricing(input_cost_per_1k=0.00015, output_cost_per_1k=0.0006)})
compute_llm_cost(pricing, "gpt-4o-mini", prompt_tokens=1000, completion_tokens=500)  # -> 0.00045
estimate_cache_savings(pricing, "gpt-4o-mini", 1000, 500, cache_hit_rate=0.4)  # -> 0.00018, a planning number, not a cache

# 5. Resource Optimization - a deterministic rightsizing recommendation from utilization data you supply
from agentic_ai.governance.finops.resource import ResourceTier, UtilizationSample, recommend_rightsizing

tiers = [ResourceTier(name="small", capacity=4, cost_per_hour=0.5), ResourceTier(name="medium", capacity=8, cost_per_hour=1.0), ResourceTier(name="large", capacity=16, cost_per_hour=2.0)]
sample = UtilizationSample(current_tier="large", avg_utilization_pct=20, peak_utilization_pct=30)
recommend_rightsizing(sample, tiers, target_headroom_pct=20.0)  # -> recommends "medium", saves $1.00/hr

# 6. Cost Forecasting & Reporting - a least-squares trend projection + a grouped report over the ledger
from agentic_ai.governance.finops.forecasting import cost_report, forecast_costs

forecast_costs([10.0, 12.0, 14.0, 16.0, 18.0], periods_ahead=3)  # -> forecast=[20.0, 22.0, 24.0], trend_per_period=2.0
cost_report(ledger, group_by="team")  # -> total_usd=5.25, breakdown={"growth": 1.25, "ml-platform": 4.00}
```

## Resilience & Reliability

One submodule per capability - `agentic_ai.resilience.<domain>`.
`high_availability`, `fault_tolerance`, `circuit_breaker`, `fallback`,
and `disaster_recovery` are new code; `retry` bridges
`agentic_ai.governance.security.runtime`'s timeout for its timeout
half; `checkpoint` bridges `agentic_ai.memory.wrappers.checkpointing`
(not duplicated - snapshot/restore is exactly what "resume an
interrupted workflow" needs); `capacity` bridges
`agentic_ai.guardrails.abuse.RateLimiter` for rate limiting. Not here
at all - genuinely not something a Python client library can do:
actually running redundant infrastructure, multi-zone/region failover,
autoscaling, or a cloud provider's own backup service. Every module
below gives you the structural primitive real resilience engineering
is built from - the infrastructure underneath stays yours to provide.

```python
# 1. High Availability - route around an unhealthy endpoint using a health check you supply (new)
from agentic_ai.resilience.high_availability import Endpoint, select_healthy_endpoint

endpoints = [Endpoint(name="primary", priority=0), Endpoint(name="secondary", priority=1)]
select_healthy_endpoint(endpoints, health_check=lambda e: ping(e))  # raises NoHealthyEndpointError if none are healthy

# 2. Fault Tolerance & Failure Handling - classify an exception so retry/circuit-breaker/fallback share one answer (new)
from agentic_ai.resilience.fault_tolerance import classify_failure

classify_failure(ConnectionError("connection reset"))  # -> FailureCategory.TRANSIENT

# 3. Retry & Timeout Management - bounded exponential backoff, never infinite (new) + a thread-based timeout (bridged)
from agentic_ai.resilience.retry import RetryPolicy, retry, check_iteration_limit, run_with_timeout

retry(call_llm, policy=RetryPolicy(max_attempts=3, base_delay_seconds=0.5))  # retries TRANSIENT/RATE_LIMITED/TIMEOUT, not PERMANENT
check_iteration_limit(step_count, max_iterations=50)  # raises MaxIterationsExceeded - the explicit agent-loop guard

# 4. Circuit Breaker & Bulkhead - isolate a failing dependency, cap concurrent calls to it (new)
from agentic_ai.resilience.circuit_breaker import Bulkhead, CircuitBreaker

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout_seconds=30.0)
bulkhead = Bulkhead(max_concurrent=10)
breaker.call(call_vector_db)  # raises CircuitOpenError immediately once the circuit trips - no wasted retries on a dependency that's down

# 5. Fallback & Recovery - fall back through a chain of models/tools/services (new)
from agentic_ai.resilience.fallback import with_fallback

with_fallback(call_primary_model, call_backup_model, args=("summarize this",))  # raises AllFallbacksFailedError only if every candidate fails

# 6. State Checkpointing & Resume - snapshot and restore agent state, the same primitive that recovers an interrupted workflow
from agentic_ai.resilience.checkpoint import checkpoint, list_checkpoints, restore

cp = checkpoint(agent_store, "workflow-42", label="before risky step")
latest = list_checkpoints("workflow-42", checkpoint_store=agent_store)[0]  # newest first - what a resume restores from
restore(agent_store, "workflow-42", latest.checkpoint_id)

# 7. Disaster Recovery - a portable file-based backup + real RPO/RTO arithmetic against an objective you define (new)
from agentic_ai.resilience.disaster_recovery import RecoveryObjective, evaluate_rpo, export_backup, import_backup

objective = RecoveryObjective(rpo_seconds=3600, rto_seconds=600)
evaluate_rpo(objective, last_backup_at)  # True if a disaster right now would lose no more than rpo_seconds of data
export_backup(agent_store, "workflow-42", "backup.json", label="nightly")  # portable - survives the live store being lost entirely
import_backup(disaster_recovery_store, "backup.json")

# 8. Capacity & Load Management - rate limiting (bridged) + a bounded queue with a real backpressure signal (new)
from agentic_ai.resilience.capacity import BoundedQueue, RateLimiter

limiter = RateLimiter(limit=100, window_seconds=60, redis_url="redis://localhost:6379/0")  # shared correctly across replicas
queue = BoundedQueue(maxsize=1000)
queue.enqueue(task)  # raises QueueFullError immediately when at capacity, rather than blocking or growing unbounded
```

## Observability

One submodule per capability - `agentic_ai.observability.<domain>`.
Built entirely on real, standard backends, not this SDK's own
invented formats: **OpenTelemetry** for both tracing and metrics,
exported to **Jaeger** (OTLP/HTTP) and **Prometheus** (via
`opentelemetry-exporter-prometheus`) respectively, with a real
**Grafana** dashboard provisioned straight from those Prometheus
metrics, and **Logs & Events persisted to Postgres**. Span names
follow a `<domain>.<action>` convention (`agent.<name>`, `llm.invoke`,
`rag.retrieval`, `tool.<name>`) so every capability's spans read
consistently in Jaeger's UI. Install with the `[observability]` extra
(or `[tracing]`/`[metrics]` individually).

```python
# 1. Distributed Tracing - real OpenTelemetry spans, exported via OTLP/HTTP to Jaeger
from agentic_ai.observability.tracing import Tracer, get_current_trace_id

tracer = Tracer("my-service", otlp_endpoint="http://localhost:4318/v1/traces")
with tracer.span("agent.job_search", attributes={"agent.step_type": "orchestration"}):
    with tracer.span("llm.invoke", attributes={"llm.model": "gpt-4o-mini"}):
        ...  # nests automatically via OTel's own context propagation - no manual trace_id plumbing

# 2. Application & Infrastructure Metrics - the OpenTelemetry Metrics API, exported in real Prometheus format
from agentic_ai.observability.metrics import MetricsRegistry

metrics = MetricsRegistry("my-service")
metrics.record_request("llm", outcome="success", latency_seconds=0.42)
body, content_type = metrics.expose()  # ready for any Prometheus server's scrape config; P95/P99 via histogram_quantile()

# 3. LLM Observability - a span + metrics per call, cost attached when a PricingTable is supplied
from agentic_ai.observability.llm_observability import ObservedLiteLLMGateway
from agentic_ai.governance.finops.pricing import ModelPricing, PricingTable

pricing = PricingTable(rates={"gpt-4o-mini": ModelPricing(input_cost_per_1k=0.00015, output_cost_per_1k=0.0006)})
observed_gateway = ObservedLiteLLMGateway(gateway, tracer=tracer, metrics=metrics, pricing=pricing)
observed_gateway.chat_completion("gpt-4o-mini", messages)  # emits llm.invoke span with tokens/latency/cost + Prometheus metrics

# 4. Agent Observability - agent steps/decisions/tool calls/loops/retries/handoffs as agent.<name> spans
from agentic_ai.observability.agent_observability import trace_agent_step

with trace_agent_step(tracer, "job_search", step_type="orchestration", metrics=metrics):
    ...

# 5. RAG Observability - retrieval latency, query, retrieved sources, top-K, failures
from agentic_ai.observability.rag_observability import trace_retrieval

with trace_retrieval(tracer, "python developer remote", top_k=5, metrics=metrics) as result:
    result["sources"] = ["doc-1", "doc-2"]  # source identifiers only, never raw retrieved text

# 6. Tool Observability - tool calls, success/failure, parameter metadata (never values), latency, downstream errors
from agentic_ai.observability.tool_observability import trace_tool_call

with trace_tool_call(tracer, "search_jobs", argument_keys=["query", "location"], metrics=metrics):
    ...

# 7. Logs & Events - application/agent/tool/security/guardrail events, correlated by trace_id, persisted to Postgres
from agentic_ai.observability.events import EventLog, EventSeverity

events = EventLog("platform", postgres_url="postgresql://user:pass@localhost:5432/observability")
events.log_event("agent", "step_started", severity=EventSeverity.INFO, trace_id=get_current_trace_id())

# 8. Alerting & Incident Detection - real threshold evaluation + z-score anomaly detection
from agentic_ai.observability.alerting import AlertRule, detect_anomaly, evaluate_alerts

rules = [AlertRule(name="high-error-rate", metric_name="error_rate", threshold=0.05)]
evaluate_alerts(rules, {"error_rate": 0.08})  # -> [Alert(triggered=True, ...)]
detect_anomaly(current_latency, baseline_latencies)  # -> AnomalyResult(is_anomaly=True/False, z_score=...)

# 9. Correlation & Root Cause Analysis - every event for one trace_id, plus a direct link into Jaeger's own UI
from agentic_ai.observability.correlation import build_jaeger_trace_url, correlate

timeline = correlate(events, trace_id)  # the full cross-component timeline for one request, oldest first
build_jaeger_trace_url("http://localhost:16686", trace_id)  # -> a one-click link to the matching span timeline

# Bonus: provision a real Grafana dashboard reading straight from the Prometheus metrics above
from agentic_ai.observability.grafana import build_observability_dashboard, provision_dashboard

dashboard = build_observability_dashboard(datasource_uid="platform-prometheus")
provision_dashboard("http://localhost:3000", dashboard, username="admin", password="admin")
```

## Monitoring & Alerting

The key difference from Observability: Observability collects the raw
signal (logs, metrics, traces). Monitoring is the engine sitting on
top of it - threshold + anomaly detection deciding when that signal
crosses a line, turned into a routed, deduplicated, escalatable alert,
and (via `incident.py`) a real incident in an on-call system. One
submodule per capability - `agentic_ai.monitoring.<domain>` - covering
the 7 core capabilities of this reference architecture: Health
Monitoring, Performance Monitoring, AI/Agent Monitoring, Quality
Monitoring, Cost Monitoring, Security/Guardrail Monitoring, and
Alerting & Incident Management. `rules.py`'s
`DEFAULT_AI_AGENT_ALERT_RULES` are concrete, ready-to-use thresholds
for an agentic AI system specifically.

```python
# 1. Health Monitoring - availability across Agent/LLM/RAG/Tool/API/DB/infra, one uniform probe shape
from agentic_ai.monitoring.health import HealthMonitor, HealthStatus

health = HealthMonitor()
health.register("llm_gateway", lambda: True)  # a bool, or a HealthStatus directly
health.register("vector_db", lambda: HealthStatus.DEGRADED)
report = health.check_all()  # overall_status is the worst individual component - one UNHEALTHY dependency makes the whole report UNHEALTHY

# 2. Performance Monitoring - latency percentiles/error rate/throughput, pure Python, no Prometheus query needed
from agentic_ai.monitoring.performance import compute_performance_summary

summary = compute_performance_summary(latency_samples, total_requests=1000, failed_requests=12, window_seconds=60)
summary.latency.p95  # same P95 a Prometheus histogram_quantile() query would give you, computed in-process

# 3. AI/Agent Monitoring - task failure/loop/retry/tool-failure/model-error rates
from agentic_ai.monitoring.agent_monitoring import AgentMonitoringCounts, compute_agent_monitoring_snapshot

counts = AgentMonitoringCounts(total_tasks=1000, failed_tasks=80, total_steps=5000, loop_detections=120, total_tool_calls=1500, tool_failures=90)
snapshot = compute_agent_monitoring_snapshot(counts)  # -> task_failure_rate=0.08, loop_rate=0.024, tool_failure_rate=0.06, ...

# 4. Quality Monitoring - a rolling window of sampled quality scores, alerting on a drop, not a single bad score
from agentic_ai.monitoring.quality_monitoring import QualityMonitor

quality = QualityMonitor(window_size=200)
quality.record(judge_score, source="llm_judge_grounding")  # scoring itself is agentic_ai.evaluation.llm's job
quality.check_drift(baseline_mean=0.9, degradation_threshold_pct=10.0)  # -> is_degraded=True/False

# 5. Cost Monitoring - agentic_ai.governance.finops re-exported, plus the specific "cost/request up 50%" signal
from agentic_ai.monitoring.cost_monitoring import detect_cost_per_request_spike

detect_cost_per_request_spike(current_cost_per_request, baseline_cost_per_request)  # -> CostSpikeResult(is_spike=True/False, increase_pct=...)

# 6. Security & Guardrail Monitoring - policy violations, prompt injection, PII, unauthorized actions, as one rate
from agentic_ai.monitoring.security_monitoring import compute_security_monitoring_snapshot

compute_security_monitoring_snapshot(guardrail_audit_events)  # -> violation_rate, prompt_injection_count, pii_count, unauthorized_count

# 7. Alerting & Incident Management - dedup/route/escalate on top of threshold+anomaly detection, then a real PagerDuty incident
from agentic_ai.monitoring.alerting import AlertManager, evaluate_alerts
from agentic_ai.monitoring.rules import DEFAULT_AI_AGENT_ALERT_RULES
from agentic_ai.monitoring.incident import PagerDutyClient

manager = AlertManager()
manager.on("critical", lambda fired: PagerDutyClient(routing_key).trigger(fired.alert.rule.name, dedup_key=fired.dedup_key, severity="critical"))

alerts = evaluate_alerts(DEFAULT_AI_AGENT_ALERT_RULES, current_metric_values)
for alert in alerts:
    manager.fire(alert, scope="tenant-acme")  # repeats increment .count instead of re-notifying; a cleared condition auto-resolves

manager.escalate(timeout_seconds=300, handler=lambda fired: page_secondary_oncall(fired))  # unacknowledged alerts past the timeout

# Bonus: publish this module's own signals into the same Prometheus/Grafana stack Observability already exports to
from agentic_ai.monitoring.telemetry import MonitoringTelemetry
from agentic_ai.monitoring.grafana import build_monitoring_dashboard, provision_dashboard

telemetry = MonitoringTelemetry(metrics)  # metrics = the same agentic_ai.observability.metrics.MetricsRegistry instance
telemetry.set_health("llm_gateway", HealthStatus.HEALTHY)
telemetry.set_rate("agent_task_failure_rate", snapshot.task_failure_rate)

dashboard = build_monitoring_dashboard(datasource_uid="platform-prometheus")  # 9 panels, threshold lines matching DEFAULT_AI_AGENT_ALERT_RULES exactly
provision_dashboard("http://localhost:3000", dashboard, username="admin", password="admin")
```

## Core Agent Capabilities

The 10-capability agent loop (Goal -> Planner -> Plan -> [Reason -> Act
-> Observe -> Reflect -> Recover]\* -> Feedback -> self-learning),
built as composable, **framework-agnostic** pieces - not a competing
orchestration framework. You don't need `AgentExecutor` to get value
here: `task`/`agent`/`tool`/`knowledge_base`/`feedback` are plain data
and plain functions, callable directly from a LangChain/LangGraph/
CrewAI/AutoGen/Semantic Kernel/Google ADK node, or a hand-rolled loop.
`AgentExecutor` is for when you want this package's own ReAct-with-
reflection-and-replan control flow, complete with checkpoint-per-step
persistence, idempotent resume, and self-learning already wired
together - async throughout (`agentic_ai.gateway`/`LLMJudge` are sync,
so sync calls run via `asyncio.to_thread` - a real wrapper, not
decorative async).

**Authentication and authorization are mandatory by default**
(`require_auth=True`) - no tool call executes without both an
authenticated actor (`agentic_ai.identity.TokenClaims`, checked for
expiry) and a context-aware authorization decision
(`agentic_ai.governance.authorization`'s OPA PDP/PEP, called with real
risk-score/business-hours context, not a bare user/action/resource
triple). An executor built without an `actor` or without an
`authorizer` **fails closed** - every action denied, nothing silently
allowed. Use `require_auth=False` only for local prototyping.

Two distinct escalation paths: when the agent genuinely can't solve a
goal (`RecoveryStrategy` exhausts retries/replans, or
`loop_escalation_threshold` repeated actions are detected), the run
closes immediately as `AgentStatus.ESCALATED` - no approve/reject
cycle, since there's nothing to approve - and the matching
`KnowledgeBase` case is marked `requires_human=True` so a similar
future request routes straight to a human. A separate, real
pending-approval gate (`AgentStatus.AWAITING_APPROVAL` via
`approval_queue`) is available infrastructure for a genuine yes/no
decision point: `APPROVED` resumes execution, `REJECTED` closes as
`FAILED` - not triggered by the default recovery flow.

```python
# Goal & Intent Understanding (#1) / Planning & Task Decomposition (#2)
from agentic_ai.core.task import Goal
from agentic_ai.core.planner import LLMPlanner, ReasoningStrategy

goal = Goal(description="Find remote Python developer jobs matching my profile")
planner = LLMPlanner(judge, strategy=ReasoningStrategy.CHAIN_OF_THOUGHT)  # or TREE_OF_THOUGHT / GRAPH_OF_THOUGHT - any reasoning pattern

# Tool Selection & Invocation (#4)
from agentic_ai.core.tool import Tool, ToolRegistry

tools = ToolRegistry()
tools.register(Tool(name="search_jobs", description="Search open job listings", fn=search_jobs))  # sync or async fn, both supported

# Reasoning & Decision Making (#3) / Goal Completion & Response (#10)
from agentic_ai.core.reasoning import LLMReasoner, LLMValidator
from agentic_ai.core.recovery import DefaultRecoveryStrategy  # Error Handling & Recovery (#9) - reuses agentic_ai.resilience.classify_failure()

# The self-learning loop: extract similar past requests as planning context, keep updating
# their hit/success/failure counts and rating every time a similar request comes through again
from agentic_ai.core.knowledge_base import KnowledgeBase
from agentic_ai.core.feedback import FeedbackStore

knowledge_base = KnowledgeBase(postgres_url="postgresql://user:pass@localhost:5432/agents")  # or in-memory/redis/sqlite/file
feedback_store = FeedbackStore(postgres_url="postgresql://user:pass@localhost:5432/agents")

# Task Execution & Orchestration (#5) / State Management (#6) / Memory Access (#7) / Observation (#8)
from agentic_ai.core.orchestrator import AgentExecutor
from agentic_ai.identity import TokenClaims

# authentication + authorization - mandatory by default (require_auth=True), fails closed without both
actor = TokenClaims(subject="agent-service-account", realm_roles=["job_search_agent"], tenant_id="acme-corp")

executor = AgentExecutor(
    planner=planner, reasoner=LLMReasoner(judge), validator=LLMValidator(judge),
    recovery_strategy=DefaultRecoveryStrategy(max_retries_per_step=2, max_replans=3),
    tool_registry=tools, knowledge_base=knowledge_base, feedback_store=feedback_store,
    actor=actor, authorizer=opa_client,  # every tool call - authenticated actor + context-aware OPA decision
    loop_escalation_threshold=3,  # 3 repeated actions -> escalate straight to a human, no more retries
    # everything else is optional - wire in exactly what your deployment needs:
    memory_store=memory_store, input_guardrails=input_guard, output_guardrails=output_guard,
    tool_policy=tool_policy, agent_security_policy=agent_security_policy,
    audit_log=audit_log, approval_queue=approval_queue, risk_register=risk_register,
    cost_ledger=cost_ledger, pricing=pricing_table, retry_policy=retry_policy, circuit_breaker=breaker,
    tracer=tracer, metrics=metrics, event_log=event_log,
    agent_monitoring_counts=agent_counts, alert_manager=alert_manager,
)

state = await executor.run(goal, memory_scope="job-search-run-1")  # AgentState is checkpointed after every step
state.status, state.result  # AgentStatus.SUCCEEDED, "Found 3 matching roles."

# crash recovery: a fresh process just resumes - already-completed steps are never re-executed
resumed = executor.resume("job-search-run-1")
state2 = await executor.run(goal, memory_scope="job-search-run-1", resume=True)  # idempotent - returns immediately if already done

# if the agent genuinely can't solve it: closes as ESCALATED (no approve/reject), KB marked requires_human
if state.status == "escalated":
    similar_next_time = knowledge_base.find_similar(goal.description)  # -> requires_human=True, routed to a human up front

# real user feedback later updates the knowledge base's rolling stats
executor.record_user_feedback(state, succeeded=True, rating=1.0, comment="great matches")
similar = knowledge_base.find_similar("Find Python jobs I can do remotely")  # -> past cases, their success rate, and average rating
```

Multi-Agent Patterns - Router, Hierarchical, Sequential, Parallel, and
Graph/Workflow, all built as composition **over** `AgentExecutor` (or
anything else exposing an async `run(goal, **kwargs)` - your own
function, a thin wrapper around any other framework's agent), not a
new execution engine:

```python
from agentic_ai.core.multi_agent import GraphWorkflow, HierarchicalAgent, ParallelWorkflow, RouterAgent, SequentialWorkflow

# Router - classify a goal, dispatch to exactly one specialist (rule-based or LLM-judge-backed)
router = RouterAgent({"billing": billing_agent, "support": support_agent}, judge=judge, descriptions={"billing": "billing questions", "support": "everything else"})

# Hierarchical - a supervisor decomposes a goal into sub-goals delegated to named workers
supervisor = HierarchicalAgent({"researcher": researcher_agent, "writer": writer_agent}, judge=judge, parallel_workers=False)

# Sequential - each step's result feeds the next step's Goal.context["previous_result"]
pipeline = SequentialWorkflow([extract_agent, transform_agent, load_agent])

# Parallel - independent steps run concurrently via asyncio.gather
fan_out = ParallelWorkflow([region_us_agent, region_eu_agent, region_apac_agent])

# Graph - a real dependency DAG, executed in topological waves (cycle detection included)
workflow = GraphWorkflow({"fetch": fetch_agent, "validate": validate_agent, "publish": publish_agent}, {"validate": ["fetch"], "publish": ["validate"]})
```

### Code Interpreter Tools (`code_interpreter`)

The same capability OpenAI's and Claude's own agents ship - read,
write, edit, search, and run any command (`sed`, `awk`, a build/test
invocation, anything) - built on a pluggable `Sandbox` boundary rather
than one fixed backend, so you pick the isolation tier a given agent
actually needs:

- **`DockerSandbox`** - a real, disposable, network-isolated container
  (the same approach OpenAI's/Claude's own interpreters use under the
  hood). Built directly on the `docker` CLI, no `docker-py` dependency.
- **`LocalSandbox`** - a directory jail plus OS resource limits (CPU
  time, memory) around a plain subprocess - shares the host kernel, so
  it's the tier a restricted shell account gives you, not a
  container's.
- **`PythonSandbox`** - Python source executed in a dedicated child
  process with restricted builtins/imports and a real timeout-kill -
  fast, but honestly **not** a security boundary against adversarial
  code (documented in its own docstring); use it for trusted/semi-
  trusted snippets, not arbitrary user input.

All three satisfy the same two-method `Sandbox` Protocol
(`resolve_path`/`run`), so `filesystem`/`search`/`shell` work
unchanged against any of them - or your own (E2B, Modal, a Firecracker
microVM, a remote execution API).

```python
from agentic_ai.core.code_interpreter.local_sandbox import LocalSandbox
from agentic_ai.core.code_interpreter.docker_sandbox import DockerSandbox
from agentic_ai.core.code_interpreter.registry import build_code_interpreter_tools
from agentic_ai.core.tool import ToolRegistry

# pick a sandbox tier - LocalSandbox for trusted-ish code, DockerSandbox for real isolation
sandbox = LocalSandbox("/tmp/agent-workspace", max_cpu_seconds=30, max_memory_bytes=1_024 * 1024 * 1024)
# sandbox = DockerSandbox("/tmp/agent-workspace", image="python:3.12-slim", network_disabled=True)

tools = ToolRegistry()
for tool in build_code_interpreter_tools(sandbox):
    tools.register(tool)  # read_file, write_file, edit_file, list_directory, grep, glob_files, code_interpreter

# registered as "code_interpreter" - the same name OpenAI's own tools API uses ({"type": "code_interpreter"})
# and runs ANY command, no allowlist inside the tool - the boundary is the sandbox itself
await tools.invoke("code_interpreter", {"command": "grep -rn 'TODO' . | sed 's/TODO/FIXME/'"})

# plug the whole set into AgentExecutor - every call already goes through its mandatory authn/authz gate
executor = AgentExecutor(planner=planner, reasoner=reasoner, validator=validator, recovery_strategy=recovery, tool_registry=tools, actor=actor, authorizer=opa_client)
```

## MCP (Model Context Protocol) Client (`mcp_client`)

Connects to **local** MCP servers (a subprocess, speaking stdio) and
**remote** MCP servers (Streamable HTTP, optionally OAuth 2.1
-protected), built strictly on the official docs this module follows:

- Build an MCP client: <https://modelcontextprotocol.io/docs/2026-07-28/develop/build-client>
- Client Best Practices: <https://modelcontextprotocol.io/docs/2026-07-28/develop/clients/client-best-practices>

Built on the official `mcp` Python SDK (`pip install "enterprise-agentic-ai-framework[mcp_client]"`,
requires `mcp>=2.0`) - this package owns connection lifecycle,
multi-server aggregation, error normalization, and `agentic_ai`
integration; it does not reimplement transport, JSON-RPC framing, or
OAuth.

```python
import os

from agentic_ai.mcp_client.config import MCPServerConfig
from agentic_ai.mcp_client.manager import MCPClientManager
from agentic_ai.mcp_client.tools import build_mcp_tools
from agentic_ai.core.tool import ToolRegistry
from agentic_ai.core.orchestrator import AgentExecutor

# one local server (a subprocess, launched over stdio) and one remote,
# bearer-token-authenticated server - mix and match freely
configs = [
    MCPServerConfig.local("fs", command="npx", args=["-y", "@modelcontextprotocol/server-filesystem", "/data"]),
    MCPServerConfig.remote("crm", url="https://mcp.example.com/mcp", bearer_token=os.environ["CRM_MCP_TOKEN"]),
]

async with MCPClientManager(configs) as manager:
    tools = ToolRegistry()
    for tool in await build_mcp_tools(manager):  # "mcp__fs__read_file", "mcp__crm__search", ...
        tools.register(tool)

    # every MCP tool call goes through AgentExecutor's SAME mandatory authn/authz gate as any other tool
    executor = AgentExecutor(
        planner=planner, reasoner=reasoner, validator=validator, recovery_strategy=recovery,
        tool_registry=tools, actor=actor, authorizer=opa_client,
    )
    state = await executor.run(goal, memory_scope="mcp-run-1")
```

**Two authentication/authorization layers, not one** - both real,
neither substitutes for the other:

1. This package's own concern: authenticating *this process* to a
   remote MCP server, per the MCP Authorization spec (OAuth 2.1 +
   PKCE) - `MCPServerConfig.bearer_token` for a static token, or
   `MCPServerConfig.oauth` (`MCPOAuthConfig` +
   `FileTokenStorage`/`InMemoryTokenStorage`) for the full
   discovery/registration/PKCE/redirect flow. A **local** (stdio)
   server deliberately gets none of this - the spec says a stdio
   server "SHOULD NOT" use OAuth, and inherits credentials from the
   launching environment instead (`MCPServerConfig.local`'s `env`).
2. `AgentExecutor`'s own concern: authenticating and authorizing the
   *acting user* for each *tool call*, MCP-sourced or not - its
   existing `actor`/`authorizer` gate, mandatory by default. MCP tools
   are plain `agentic_ai.core.tool.Tool` objects for exactly this
   reason - no special-casing, an MCP tool call is gated the same way
   a `code_interpreter` call already is.

```python
from agentic_ai.mcp_client.oauth import MCPOAuthConfig, FileTokenStorage

remote_with_oauth = MCPServerConfig.remote(
    "crm", url="https://mcp.example.com/mcp",
    oauth=MCPOAuthConfig(client_name="my-agent", scope="tools:read tools:write", storage=FileTokenStorage("~/.agentic-ai/crm-mcp-tokens.json")),
)
```

**Client Best Practices**, followed explicitly:

- **Progressive tool discovery** - once the connected catalog is large
  enough to threaten the context window (auto-detected by default, or
  force with `progressive_discovery=True`), `build_mcp_tools()`
  registers exactly three meta-tools instead of one per MCP tool -
  `mcp_search_tools` (keyword search across every server) ->
  `mcp_get_tool_details` (full schema for one match) -> `mcp_call_tool`
  (invoke it) - the catalog/inspect/execute three-layer pattern from
  the docs above, instead of hundreds of tool definitions in every
  prompt.
- **Refresh on `list_changed`** - `MCPServerConfig(..., watch_for_changes=True)`
  opens a `subscriptions/listen` stream and invalidates the cached
  tool list on `notifications/tools/list_changed`, instead of polling
  `tools/list` before every call.
- **Dynamic server management** - `manager.connect_all(strict=False)`
  keeps whatever servers are reachable instead of failing the whole
  agent because one optional server is down (`manager.connection_failures`
  reports which, and why).
- **Correct failure semantics** - a failing MCP tool call is a
  *successful* JSON-RPC response with `is_error=True`, not a transport
  failure, per the spec. This module converts that into a raised
  `MCPToolError` so `AgentExecutor`'s Reflect/Recover step actually
  engages (retry/replan/escalate), instead of silently treating a
  failed tool as a successful Observation.

```python
# many servers at once, qualified names avoid any cross-server collision
async with MCPClientManager([local_fs_config, local_git_config, remote_crm_config]) as manager:
    await manager.connect_all(strict=False)  # keep whichever servers are reachable
    result = await manager.call_qualified_tool("mcp__crm__search", {"query": "acme corp"})

    # resources and prompts, straight from one connection
    connection = manager.connection("crm")
    doc = await connection.read_resource("crm://accounts/acme")
    messages = await connection.get_prompt("draft_followup", {"account": "acme"})
```

## Requirements

- Python 3.10+
- A LiteLLM proxy you deploy yourself (this library is a client, not a
  bundled server)
- For Memory & State: nothing extra for in-memory/file/SQLite; Redis,
  Postgres (+pgvector), or Qdrant only if you choose those backends
- For Secrets Management: a HashiCorp Vault instance you deploy
  yourself, and the `[vault]` extra
- For Guardrails: nothing extra beyond the base install for
  deterministic detectors; the LLM-judged detectors need a gateway,
  same as Evaluation
- For Authentication: a Keycloak instance you deploy yourself, and the
  `[authentication]` extra
- For Identity: nothing extra - pure data models, no optional
  dependency at all
- For Governance: an Open Policy Agent instance you deploy yourself;
  no new extra - built on the base `httpx` dependency
- For Resilience & Reliability: nothing extra beyond the base install
- For Observability: a Jaeger instance (OTLP/HTTP ingest) for tracing
  and the `[tracing]` extra; a Prometheus server to scrape metrics and
  the `[metrics]` extra (or `[observability]` for both); a Postgres
  database for Logs & Events; a Grafana instance with a Prometheus
  datasource already configured, for the optional dashboard
  provisioning
- For Monitoring & Alerting: nothing extra for the core rate/threshold/
  anomaly logic - built on the base install; the `[observability]`
  extra only if you use `telemetry.py`/`grafana.py`'s live-dashboard
  path; a PagerDuty account (and one of its services' Events API v2
  routing key) for Incident Integration
- For Core Agent Capabilities: nothing extra beyond the base install
  for `task`/`agent`/`tool`/`knowledge_base`/`feedback`/`multi_agent`;
  `AgentExecutor`'s default `LLMPlanner`/`LLMReasoner`/`LLMValidator`
  need a `LiteLLMGateway`, same as `evaluation`. For `tools` (the code
  interpreter): nothing extra for `LocalSandbox`/`PythonSandbox`
  (stdlib `subprocess`/`resource`/`multiprocessing` only); `DockerSandbox`
  needs the `docker` CLI installed and a running daemon - no Python
  package dependency, built directly on the CLI

## License

Apache-2.0
