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.

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

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, Authentication, Identity, and Governance 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.compliance, governance.data_governance, governance.finops, governance.responsible_ai	⏳ Planned
observability - distributed tracing, structured audit	⏳ Planned
monitoring, resilience	⏳ 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):

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

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:

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

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

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

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

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:

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:

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:

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

6. LLM Audit

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

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

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:

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:

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

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:

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:

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:

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:

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:

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:

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

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:

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

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

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

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

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.

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

Connecting

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

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

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:

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

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:

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:

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

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:

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:

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

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.

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

Connecting and authenticating

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

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

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:

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

Auto-refresh

The tedious part of every OAuth client, done once:

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

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.

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:

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.

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.

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

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:

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

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

Auditability

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

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:

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.

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

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

# 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"},
)

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

License

Apache-2.0