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

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

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

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.

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

Connecting and authenticating

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

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

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:

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:

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:

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

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