Metadata-Version: 2.4
Name: enterprise-agentic-ai-framework
Version: 0.8.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: dev
Requires-Dist: hvac>=2.0; extra == 'dev'
Requires-Dist: psycopg[binary]>=3.1; 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: identity
Requires-Dist: python-keycloak>=4.0; extra == 'identity'
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: 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: 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
```

## Status

This is an early release. **The LLM gateway, the full evaluation suite,
Memory & State, Context Engineering, Secrets Management, Guardrails &
Content Safety, and Identity 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 |
| `identity` - authentication: Keycloak (OIDC grants, token validation, auto-refresh) (see below) | ✅ Implemented |
| `governance` - authorization (PEP/PDP) | ⏳ Planned |
| `observability` - distributed tracing, structured audit | ⏳ Planned |
| `finops` - LLM cost tracking | ⏳ Planned |
| `security` - rate limiting, abuse detection (live enforcement) | ⏳ Planned |
| `compliance`, `audit`, `data_governance` | ⏳ Planned |
| `monitoring`, `resilience`, `responsible_ai` | ⏳ Planned |
| `core` - agent/tool base classes, orchestrator | ⏳ Planned |

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

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

## Identity

A client for Keycloak's OpenID Connect endpoints - authentication,
not authorization: it proves who someone (or some service) is and
hands back their roles/claims as data. Enforcing what those roles are
allowed to do is the future `governance` module's job, same boundary
guardrails' tool allow-listing already draws.

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

### Connecting and authenticating

```python
from agentic_ai.identity 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
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.

### Validating a token

```python
claims = client.decode_token(tokens.access_token)   # local, offline - verifies signature against Keycloak's JWKS
print(claims.subject, claims.username, claims.realm_roles)
print(claims.has_realm_role("admin"), claims.has_client_role("my-service", "editor"))

status = client.introspect(tokens.access_token)      # asks Keycloak directly - catches server-side revocation decode_token() can't see
print(status.active)

info = client.get_user_info(tokens.access_token)     # the userinfo endpoint
```

### Refresh, logout, and the authorization_code flow

```python
refreshed = client.refresh(tokens.refresh_token)
client.logout(tokens.refresh_token)

url = client.build_authorization_url("https://app.example.com/callback", state="xyz")
# ... redirect the user's browser to `url`, 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")
```

### 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

claims = client.decode_token(tokens.access_token)
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.

### Auto-refresh

The tedious part of every OAuth client, done once:

```python
from agentic_ai.identity 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.identity.wrappers import AuditedKeycloakClient

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

## 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 Identity: a Keycloak instance you deploy yourself, and the
  `[identity]` extra

## License

Apache-2.0
