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

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

Status

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

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

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

Prerequisites

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

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

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

Compliance

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

from agentic_ai.governance.compliance import ComplianceRegister, ComplianceStatus

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

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

Responsible AI

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

FinOps

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

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

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

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

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

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

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

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

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

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

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

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

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

Resilience & Reliability

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Observability

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Monitoring & Alerting

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Core Agent Capabilities

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Code Interpreter Tools (code_interpreter)

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

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

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

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

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

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

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

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

MCP (Model Context Protocol) Client (mcp_client)

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

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

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

import os

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

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

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

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

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

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

from agentic_ai.mcp_client.oauth import MCPOAuthConfig, FileTokenStorage

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

Client Best Practices, followed explicitly:

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

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

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

Requirements

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

License

Apache-2.0