An action firewall for MCP – OAuth says who, not whether the call should run

Source and examples · Framework integration guides · MCP integration guide · Issue tracker · Security policy · PyPI package

Production-grade security guardrails, DLP, tool authorization, and cryptographically verified multi-agent delegation for Python autonomous AI runtimes.

agenticdome-python-sdk

is the official Python SDK and middleware package for AgenticDome. It enforces deterministic security controls at every boundary your agents cross — prompt ingress, tool execution, agent-to-agent handoffs, and output egress — using the tenant's assigned AgenticDome runtime sidecar.

One security pattern, fifteen runtimes: CrewAI · PydanticAI · LangGraph/LangChain · Microsoft Agent Framework · Microsoft AutoGen · Microsoft AI Foundry · OpenAI Agents SDK · Claude Agent SDK · Hugging Face smolagents · Agno · Google ADK · LlamaIndex · AWS Bedrock · MCP hosts/gateways · custom Python.

pip install agenticdome-python-sdk
export AGENTICDOME_MODE=local_sim

No account, API key, tenant, network connection, or third-party framework package is required.

agenticdome-demo --framework langgraph --scenario both

Look for ALLOWED — TOOL WOULD EXECUTE

followed by BLOCKED — TOOL WOULD NOT EXECUTE

.

Offline demonstration—not runtime evidence.This command evaluates two fixed onboarding scenarios with a deterministic, bundled public baseline. It does not contact AgenticDome, load tenant policy, execute either tool, or instantiate a LangGraph graph.--framework langgraph

labels the example payload and points you to the matching integration; it is not a LangGraph integration test.

agenticdome-demo --list-frameworks
agenticdome-demo --framework all --scenario both

You can also open the framework example gallery and run the individual example matching your stack. Before editing production code, choose the dedicated framework integration guide, then use the production integration playbook for the cross-framework attachment and proof checklist. MCP host and gateway developers can follow the dedicated MCP Action Firewall guide.

When you are ready to test real tenant policy, remove AGENTICDOME_MODE=local_sim

, obtain the assigned runtime sidecar URL, Runtime/SDK API key and tenant ID from AgenticDome, and configure:

unset AGENTICDOME_MODE
export AGENTICDOME_API_BASE="https://your-assigned-sidecar.example"
export AGENTICDOME_API_KEY="your_runtime_sdk_key"
export AGENTICDOME_TENANT_ID="your_tenant_id"
agenticdome-demo --framework langgraph --scenario both --live

With --live

, the same fixed scenarios are sent to the actual assigned AgenticDome runtime sidecar, so their verdicts come from the customer's tenant policy and engine. The live demo still does not instantiate LangGraph or prove that an application has attached the framework adapter at every execution boundary. Use the framework guide to attach the adapter, test the real application path, and use AgenticDome Runtime Assurance for production evidence.

import agenticdome_sdk.crewai   # registers global CrewAI security hooks

crew = Crew(agents=[manager, specialist], tasks=[task])
result = crew.kickoff()          # hostile prompts, unsafe tools, and rogue

Five-minute developer trialWhy AgenticDomeHow It WorksQuickstartInstallationConfigurationChoosing Your Integration Point—dedicated framework guidesFramework IntegrationsCore SDK Client (Custom Runtimes)Production DeploymentSource Installation and VerificationLicensing

An agent can be hijacked while holding valid tokens, approved tools, and fully authorized paths. Legacy security sees a compliant request; the business sees a breach. AgenticDome adds an intent-aware enforcement layer at the exact boundaries where agents act:

Give every agent and workload a distinct principal and keep your identity provider authoritative for authentication, scopes, conditional access, credential lifecycle, and downstream entitlements. Then pass the identity the application has actually authenticated into AgenticDome alongside the agent, purpose, tool, final arguments, session, and delegation context. A valid token proves access authority; it does not prove that a prompt-influenced action is appropriate. AgenticDome complements IAM by evaluating that action at the application-controlled execution boundary. It does not issue enterprise identities or make a stolen credential safe.

For the architectural rationale, see An agent principal proves who—not why.

Control	What it stops	Where it runs
Prompt ingress guardrails
Prompt injection, jailbreaks, system-prompt extraction, instruction override, policy bypass	Before agent/LLM execution
Tool & skill authorization
Unauthorized or out-of-policy tool calls, evaluated on agent identity, tool name, arguments, session, source metadata, and delegation chain	Before tool execution
Cryptographic delegation handoffs
Confused-deputy attacks, lateral privilege escalation, stolen-token replay between agents	At every manager→specialist handoff
Inline output DLP
Leakage of PII, emails, phone numbers, API keys, access tokens, cloud credentials, corporate secrets, compliance-sensitive records	Before output is persisted, displayed, or re-enters the agent loop
Fail-safe runtime behavior
Silent security bypass when the assigned runtime is unavailable	Configurable fail-closed (production) / fail-open (dev)

Delegation tokens are the differentiator. Every authorized handoff issues a decision token that binds the originating human subject (when present), the ordered and nested agent actor chain, source manager and target specialist, exact tool name and arguments, session ID, trace and lineage root/parent IDs, intent digest, policy binding, authorized scopes, trust epochs, tenant context, and the policy decision itself. Execution verification consumes the server-side call budget and checks revocation state — a token authorizes one exact action, once.

Optional proof-of-possession hardens this further: a DPoP-style signed JWT is tied to the decision-token hash, so sending a public-key thumbprint alone is never treated as proof.

pip install "agenticdome-python-sdk[pop]"
python
from agenticdome_sdk import create_dpop_proof, generate_rsa_proof_key

proof_key = generate_rsa_proof_key()
proof = create_dpop_proof(
    proof_key["private_key_pem"],
    access_token=decision_token,
    method="POST",
    uri="/a2a",
)
client.a2a_verify_decision_token_rpc(
    decision_token,
    tool_name=tool_name,
    tool_args=tool_args,
    agent_id=worker_id,
    source_agent_id=manager_id,
    proof_token=proof,
)

AgenticDome uses a hybrid split-plane architecture. Your application executes agents, tools and workflows. The SDK protects the application-controlled boundaries and sends live policy checks to the tenant's assigned runtime sidecar. The management console distributes configuration to that runtime out of band; it is not the per-action SDK endpoint.

Management console / control plane
        |
        | policy and tenant configuration (out of band)
        v
Tenant-assigned runtime sidecar <------ live policy checks ------+
        |                                                        |
        +---------------- verdict / authorization -------------->|
                                                                 |
Customer application                                             |
  user input -> agent -> tool / MCP -> agent -> output            |
                 ^         ^                    ^                  |
                 +---------+--------------------+------------------+
                     AgenticDome SDK protection boundaries

The SDK must be attached in application code; setting environment variables alone does not intercept a framework. The assigned sidecar authenticates the tenant and evaluates live requests. Tools still execute in the customer's environment or selected provider unless a separate execution service has been deliberately configured.

Managed service: AgenticDome assigns the tenant a managed sidecar in the selected supported geographic region, subject to availability and the customer's plan or contract.Sovereign deployment: the runtime is deployed within the contracted customer-controlled boundary, such as a dedicated VPC, customer cloud, or on-premises environment.

The SDK does not choose or change runtime placement; it connects to the tenant-specific API base supplied during onboarding. See Runtime location and Redis responsibilities for the deployment boundary and customer responsibilities.

Persona / component	Responsibilities	Commercial model
Enterprise / organization
Hosts the local agent runtime. Uses the AgenticDome console to create policies, obtain a Tenant ID, generate API keys, and monitor security events.	Paid subscriber (SaaS license or API volume)
Agent / tool developer
Builds tools, skills, agents, and workflow components. Uses the SDK to support secure tool calls, delegation metadata, and DLP-aware outputs.	Free ecosystem partner — no subscription required
This Python SDK
Runs inside the application process. Protects supported framework boundaries, calls the assigned runtime sidecar, and enforces returned policy results.	Runtime security utility
Assigned runtime sidecar
Authenticates the tenant and evaluates live guardrail, tool, delegation and output requests using distributed policy.	Runtime enforcement service
Management console / control plane
Manages tenant configuration, policy distribution, governance workflows and evidence. It is not the SDK's per-action API URL.	Management plane

Start with a network-free simulation, then connect the same SDK to your assigned runtime sidecar for real tenant enforcement.

1. Try the installed local simulation. No account, API key, tenant, network call, telemetry, or framework package is required:

pip install agenticdome-python-sdk
agenticdome-demo --framework langgraph --scenario both

agenticdome-demo --framework all --scenario both
agenticdome-demo --list-frameworks

This is visibly labelled LOCAL SIMULATION — NOT CLOUD ENFORCEMENT. It evaluates two fixed inputs with a small deterministic bundled baseline through the public core-client response shape. The --framework

option changes the payload label and integration guidance; it does not import, instantiate, or run the selected framework. The simulation does not load tenant policy, issue signed decision tokens or execution receipts, write cloud evidence, or provide runtime assurance. To exercise a real adapter inside your own process without credentials, set AGENTICDOME_MODE=local_sim

in that application or framework example. The SDK refuses that mode when AGENTICDOME_PRODUCTION_MODE=true

.

Browse the public examples/ gallery for a runnable allowed/blocked example for CrewAI, PydanticAI, LangGraph, Microsoft Agent Framework, AutoGen, AI Foundry, OpenAI Agents, Claude, smolagents, Agno, Google ADK, LlamaIndex, Bedrock, MCP, and custom Python. Local blocked/redacted results emit safe terminal logs containing verdict metadata only—not raw prompts, arguments, keys, or secrets.

2. Onboard for real enforcement. Create an account in the AgenticDome Management Console, obtain your tenant identifier and Runtime / SDK API key, and identify the tenant's assigned runtime sidecar.

3. Install the SDK with the extra matching your framework:

pip install "agenticdome-python-sdk[crewai]"     # or pydanticai, langgraph, ...

4. Configure the three required production environment variables (live clients fail with a configuration error rather than silently running unprotected):

export AGENTICDOME_API_BASE="https://demo-sidecar.agenticdome.io"
export AGENTICDOME_API_KEY="your_api_key_abc123..."
export AGENTICDOME_TENANT_ID="your_tenant_id_xyz789..."

5. Attach AgenticDome at your framework's boundary — environment variables alone never intercept execution; every framework needs its one-time code attachment (see Choosing Your Integration Point):

import agenticdome_sdk.crewai

from crewai import Crew
crew = Crew(agents=[manager, specialist], tasks=[task])
result = crew.kickoff()

6. Verify locally or against the real assigned sidecar:

agenticdome-demo --framework crewai --scenario both
agenticdome-demo --framework langgraph --scenario both
agenticdome-demo --framework claude --scenario both
agenticdome-demo --framework smolagents --scenario both
agenticdome-demo --framework all --scenario both

agenticdome-demo --framework claude --scenario metadata_exfil
agenticdome-demo --framework smolagents --scenario metadata_exfil

agenticdome-demo --framework langgraph --scenario safe_lookup --live

Every Python integration listed below is selectable through agenticdome-demo --framework ...

; the demo prints the correct package extra and integration import for that framework. Local simulation proves SDK compatibility and control flow only. The live command proves the configured tenant-sidecar path.

Install the core SDK alone for custom runtimes, or add the extra for your framework:

pip install agenticdome-python-sdk

Target runtime	Command
CrewAI	pip install "agenticdome-python-sdk[crewai]"
PydanticAI	pip install "agenticdome-python-sdk[pydanticai]"
LangGraph / LangChain	pip install "agenticdome-python-sdk[langgraph]"
Microsoft Agent Framework	pip install "agenticdome-python-sdk[microsoft]"
Microsoft AutoGen AgentChat / Core (Python 3.10+)	pip install "agenticdome-python-sdk[autogen]"
Microsoft AI Foundry	pip install "agenticdome-python-sdk[foundry]"
OpenAI Agents SDK	pip install "agenticdome-python-sdk[openai-agents]"
Anthropic Claude Agent SDK	pip install "agenticdome-python-sdk[claude]"
Hugging Face smolagents	pip install "agenticdome-python-sdk[smolagents]"
Agno	pip install "agenticdome-python-sdk[agno]"
Google ADK	pip install "agenticdome-python-sdk[google-adk]"
LlamaIndex	pip install "agenticdome-python-sdk[llamaindex]"
AWS Bedrock / Bedrock Agents	pip install "agenticdome-python-sdk[bedrock]"
MCP host / gateway	pip install "agenticdome-python-sdk[mcp]"
Optional cross-process delegation store	pip install "agenticdome-python-sdk[redis]"
Proof-of-possession helpers	pip install "agenticdome-python-sdk[pop]"
All optional integrations	pip install "agenticdome-python-sdk[all]"

Some adapters are dependency-light at import time: Google ADK, LlamaIndex, Bedrock, MCP, and Microsoft helpers can wrap local boundaries without forcing one exact runtime stack. Install the framework packages your application actually uses.

AgenticDome supports framework versions inside the dependency ranges declared by the published package. Check the package metadata and compatibility table before upgrading an integration, especially for multi-package stacks such as LangGraph/LangChain.

This means upgrading support for the latest framework does not silently drop customers on the previously certified version. Versions below the displayed certified floor are not claimed as supported until they are tested. Customers managing framework dependencies themselves may install the core SDK without an extra, but their framework version must still be inside the certified range for a production support claim.

AgenticDome has two setup layers, and both are required:

Configuration layer— environment variables that every framework reads at runtime. Global to the process, container, worker, or serverless function.Code integration layer— the framework boundary where AgenticDome is attached, imported, or wrapped (next section).

Environment variables do not intercept framework execution by themselves.

Every supported Python integration inherits the same credential-free, network-free simulator:

export AGENTICDOME_MODE="local_sim"

No API base, API key, tenant ID, or third-party framework package is needed to run the public demonstration. ALLOWED

decisions are logged at INFO

; BLOCKED

and REDACTED

decisions are logged at WARNING

so they are visible in a normal terminal. Logs contain decision metadata only and exclude raw prompt text and tool arguments.

Local simulation is deliberately limited: it uses the bundled public baseline, does not execute tools, does not use tenant policy, topology, signed provenance, runtime telemetry, decision tokens, or execution receipts, and cannot satisfy an enforced execution broker. It is refused whenever AGENTICDOME_PRODUCTION_MODE=true

.

export AGENTICDOME_API_BASE="https://demo-sidecar.agenticdome.io"   # tenant runtime sidecar, not the console URL
export AGENTICDOME_API_KEY="your_api_key_abc123..."
export AGENTICDOME_TENANT_ID="your_tenant_id_xyz789..."
export AGENTICDOME_FAIL_CLOSED="true"
export AGENTICDOME_REDACT_PII="true"
export AGENTICDOME_REDACT_SECRETS="true"
export AGENTICDOME_BLOCK_ON_SENSITIVE_OUTPUT="false"
export AGENTICDOME_REQUIRE_TOKEN="true"
export AGENTICDOME_REPORT_INCIDENTS="true"

Redis is not required for normal SDK policy calls or when using an assigned managed sidecar. Add customer-managed Redis only when manager-to-specialist delegation is authorised in one application process, worker, or pod and its one-time handoff state must be consumed in another:

export AGENTICDOME_REDIS_URL="redis://redis.internal:6379/0"
export AGENTICDOME_REDIS_KEY_PREFIX="AgenticDome:production:handoff"

Runtime style	Put AgenticDome config here
Local development	Shell exports, .env , direnv, or your IDE run configuration
Docker / Compose	environment: entries, env_file: , or secret-mounted environment variables
Kubernetes	Secret / ConfigMap values injected into the deployment or job
CI/CD workers	Pipeline secret variables
Celery / RQ / background workers	Worker process environment — not only the web process
Serverless	Function environment variables or secret manager bindings

export AGENTICDOME_PLATFORM="crewai"                  # runtime platform label used in policy context
export AGENTICDOME_REQUIRE_SESSION_ID="false"         # require explicit session IDs (vs fallback local IDs)
export AGENTICDOME_DEFAULT_TOOL_PLATFORM="unknown"    # fallback platform for tools
export AGENTICDOME_HANDOFF_TOKEN_TTL_S="900"          # delegation token lifetime in seconds
export AGENTICDOME_BLOCKED_INCIDENT_SEVERITY="medium" # default severity for incident reports
export AGENTICDOME_PRODUCTION_MODE="false"            # production hardening (stable session ID enforcement)

Full configuration reference — core variables

Environment Variable	Type	Default	Description
AGENTICDOME_API_BASE
string	required	Tenant runtime sidecar origin, e.g. https://demo-sidecar.agenticdome.io . Separate from the control-plane console URL.
AGENTICDOME_API_KEY
string	required	API key generated in the AgenticDome console.
AGENTICDOME_TENANT_ID
string	required	Tenant or organization isolation namespace.
AGENTICDOME_MODE
live / local_sim
live
Selects real sidecar enforcement or the credential-free, network-free demonstration evaluator. local_sim is refused in production mode.
AGENTICDOME_PLATFORM
string	framework-specific	Runtime platform label included in policy context.
AGENTICDOME_TIMEOUT_S
integer	20
HTTP timeout in seconds for SDK calls.
AGENTICDOME_FAIL_CLOSED
boolean	true
Blocks execution if security checks fail.
AGENTICDOME_REDACT_PII
boolean	true
Enables PII redaction for output review.
AGENTICDOME_REDACT_SECRETS
boolean	true
Enables secret and credential redaction.
AGENTICDOME_BLOCK_ON_SENSITIVE_OUTPUT
boolean	false
Blocks entire output when sensitive content is detected.
AGENTICDOME_REQUIRE_TOKEN
boolean	true
Requires delegated specialist executions to include a token.
AGENTICDOME_REQUIRE_SESSION_ID
boolean	framework-specific	Requires explicit session ID for strict audit mapping.
AGENTICDOME_DEFAULT_TOOL_PLATFORM
string	unknown / python
Fallback platform for tools.
AGENTICDOME_HANDOFF_TOKEN_TTL_S
integer	900
Delegation token TTL in seconds.
AGENTICDOME_REDIS_URL
string	empty	Optional customer-application Redis URL, needed only when one-time delegation state must cross processes, workers, or pods. It is unrelated to the managed sidecar's internal backing services.
AGENTICDOME_REDIS_KEY_PREFIX
string	framework-specific	Optional key prefix for the customer application's cross-process delegation store.
AGENTICDOME_TOKEN_HMAC_SECRET
string	empty	Optional secret-manager value used by the SDK to protect shared delegation state. Applications should not inspect or construct that state.
AGENTICDOME_PRODUCTION_MODE
boolean	false
Enables production hardening such as stable session ID enforcement.
AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD
boolean	true
Requires a stable session/run/trace ID when production mode is enabled.
AGENTICDOME_CLOUD_PROVIDER
string	empty	Optional cloud/provider label added to policy context.
AGENTICDOME_CLOUD_PROJECT_ID
string	empty	Optional project/account label added to policy context.
AGENTICDOME_IDENTITY_PROVIDER
string	empty	Optional identity-provider label added to policy context.
AGENTICDOME_ENABLE_COPILOT_THREAT_API
boolean	false
Enables optional Microsoft Copilot / AI Foundry threat helper calls where available.
AGENTICDOME_ENFORCE_COPILOT_THREAT_API
boolean	false
Makes optional Copilot / AI Foundry threat helper failures or blocks enforce locally.
AGENTICDOME_COPILOT_API_VERSION
string	2025-09-01
API version used by optional Copilot / AI Foundry threat helper calls.
AGENTICDOME_BEARER_TOKEN
string	optional	Bearer token used by Microsoft AI Foundry threat-contract endpoints.
AGENTICDOME_REPORT_INCIDENTS
boolean	true
Reports blocked actions and middleware failures.
AGENTICDOME_BLOCKED_INCIDENT_SEVERITY
string	medium
Default severity for incident reports.

Full configuration reference — per-adapter variables

Every framework adapter exposes the same family of local hardening controls, prefixed per adapter: AGENTICDOME_CREWAI_*

, AGENTICDOME_PYDANTICAI_*

, AGENTICDOME_LANGGRAPH_*

, AGENTICDOME_MSAF_*

(Microsoft Agent Framework), AGENTICDOME_AUTOGEN_*

, AGENTICDOME_FOUNDRY_*

, AGENTICDOME_OPENAI_AGENTS_*

, AGENTICDOME_CLAUDE_*

, AGENTICDOME_SMOLAGENTS_*

, AGENTICDOME_AGNO_*

, AGENTICDOME_BEDROCK_*

, AGENTICDOME_GOOGLE_ADK_*

, and AGENTICDOME_MCP_*

.

Common per-adapter pattern (substitute the prefix for your adapter):

Variable suffix	Type	Default	Description
_MAX_INPUT_CHARS
integer	50000
Maximum prompt/input text reviewed before local truncation or blocking.
_MAX_OUTPUT_CHARS
integer	100000
Maximum output text reviewed before local truncation or blocking.
_MAX_TOOL_ARG_CHARS
integer	20000
Maximum serialized tool arguments before blocking.
_STREAMING_BUFFER_CHARS
integer	4000
Sliding buffer used by streaming sanitization helpers (CrewAI, Agno, OpenAI Agents, Bedrock, Google ADK, LangGraph).
_RATE_LIMIT_PER_MINUTE
integer	0
Per-agent/session/purpose local rate limit; 0 disables it.
_RETRY_ATTEMPTS
integer	2
Retry attempts for policy client calls.
_RETRY_BACKOFF_S
float	0.25
Initial exponential backoff delay for policy client retries.
_CIRCUIT_BREAKER_FAILURES
integer	5
Consecutive policy call failures before opening the local circuit breaker.
_CIRCUIT_BREAKER_RESET_S
integer	60
Seconds before retrying after the circuit breaker opens.
_AUDIT_LOGGING
boolean	true
Emits structured audit logs from the adapter.
_OTEL_ENABLED
boolean	true
Emits OpenTelemetry span events when OpenTelemetry is installed and a span is active.
_EMERGENCY_BLOCK_TOOLS
CSV string	empty	Local emergency deny list for tool names.
_EMERGENCY_BLOCK_AGENTS
CSV string	empty	Local emergency deny list for agent IDs.

Adapter-specific additions:

Environment Variable	Type	Default	Description
AGENTICDOME_LANGGRAPH_AGENT_ID
string	langgraph_orchestrator
Default LangGraph orchestrator node identity.
AGENTICDOME_LANGGRAPH_FINAL_ID
string	langgraph_final_node
Default LangGraph final-output node identity.
AGENTICDOME_LANGGRAPH_REQUIRE_SERVER_TOKENS
boolean	false
Requires handoff authorization responses to include server-issued decision tokens.
AGENTICDOME_CLAUDE_AGENT_ID
string	claude_agent
Default Claude Agent SDK identity.
AGENTICDOME_CLAUDE_STRICT_DELEGATED_EXECUTION
boolean	true
Requires server-issued decision tokens for Claude multi-agent handoffs.
AGENTICDOME_SMOLAGENTS_AGENT_ID
string	smolagent
Default smolagents identity.
AGENTICDOME_SMOLAGENTS_SCAN_CODE_EXPRESSIONS
boolean	true
Reviews CodeAgent-generated Python immediately before executor invocation.
AGENTICDOME_SMOLAGENTS_STRICT_DELEGATED_EXECUTION
boolean	true
Authorizes and verifies managed-agent handoffs using bound decision tokens.
AGENTICDOME_LANGGRAPH_STRICT_DELEGATED_EXECUTION
boolean	true
Blocks delegated executions that carry delegation metadata without a valid token.
AGENTICDOME_FOUNDRY_REQUIRE_OUTPUT_SANITIZATION_IN_PROD
boolean	true
Requires API-key-backed Mesh output sanitization when Foundry production mode is enabled.
AGENTICDOME_BEDROCK_AGENT_ID
string	aws_bedrock_agent
Default agent identity for Bedrock runtime calls and local action handlers.
AGENTICDOME_BEDROCK_MODEL_ID
string	empty	Optional default Bedrock model ID for policy context.
AGENTICDOME_AWS_ACCOUNT_ID
string	empty	AWS account ID added to Bedrock policy context when available.
AGENTICDOME_AWS_REGION
string	AWS_REGION / AWS_DEFAULT_REGION
AWS region added to Bedrock policy context when available.
AGENTICDOME_AWS_ROLE_ARN
string	empty	AWS role ARN added to Bedrock policy context.
AGENTICDOME_AWS_PRINCIPAL_ARN
string	empty	AWS principal/caller ARN added to Bedrock policy context.
AGENTICDOME_SANITIZE_MODEL_OUTPUT
boolean	true
Enables Mesh output review for model responses before returning to the application.
AGENTICDOME_GOOGLE_ADK_AGENT_ID
string	google_adk_agent
Default agent identity for Google ADK callback enforcement.
AGENTICDOME_LLAMAINDEX_AGENT_ID
string	llamaindex_agent
Default agent identity for LlamaIndex tools, query engines, and retrievers.
AGENTICDOME_SANITIZE_QUERY_OUTPUT
boolean	true
Enables Mesh output review for LlamaIndex query responses.
AGENTICDOME_MCP_HOST_ID
string	MCP_Enterprise_Host
Default agent identity for the MCP host or gateway process.
AGENTICDOME_MCP_TOOL_PLATFORM
string	mcp_third_party_server
Default downstream MCP server platform label for policy and billing context.
AGENTICDOME_SANITIZE_TOOL_OUTPUT
boolean	true
Enables Mesh output review for tool results before returning to the client.
AGENTICDOME_SANITIZE_RESOURCE_OUTPUT
boolean	true
Enables Mesh output review for MCP resource read results.
AGENTICDOME_SANITIZE_PROMPT_OUTPUT
boolean	true
Enables Mesh output review for MCP prompt results.
AGENTICDOME_SANITIZE_STREAMING_OUTPUT
boolean	true
Enables chunk-level sanitization helpers for streaming MCP responses.
AGENTICDOME_VERIFY_DECISION_TOKENS
boolean	true
Verifies delegated decision tokens when MCP tool calls carry handoff metadata.
AGENTICDOME_SCREEN_UPSTREAM_PROMPT
boolean	true
Screens upstream user prompt text in MCP host context before tool forwarding.
AGENTICDOME_MCP_PROTECT_TOOLS_LIST
boolean	true
Authorizes and filters MCP tools/list discovery responses.
AGENTICDOME_MCP_PROTECT_RESOURCES_LIST
boolean	true
Authorizes MCP resources/list discovery requests.
AGENTICDOME_MCP_PROTECT_RESOURCES_READ
boolean	true
Authorizes MCP resources/read requests before forwarding.
AGENTICDOME_MCP_PROTECT_PROMPTS_LIST
boolean	true
Authorizes MCP prompts/list discovery requests.
AGENTICDOME_MCP_PROTECT_PROMPTS_GET
boolean	true
Authorizes MCP prompts/get requests before forwarding.
AGENTICDOME_MCP_PROTECT_SAMPLING_CREATE_MESSAGE
boolean	true
Authorizes MCP sampling/createMessage requests.
AGENTICDOME_MCP_SERVER_ID
string	empty	Default MCP server identity included in policy context.
AGENTICDOME_MCP_SERVER_URL
string	empty	Default MCP server URL included in policy context.
AGENTICDOME_MCP_SERVER_TRUST_LEVEL
string	empty	Default MCP server trust label included in policy context.
AGENTICDOME_MCP_SERVER_VENDOR
string	empty	Default MCP server vendor included in policy context.
AGENTICDOME_MCP_MAX_REQUEST_TEXT_CHARS
integer	20000
Maximum upstream request text sent for prompt/method authorization before local truncation.

One table, one decision. Find your runtime, apply the required code action, and jump to its guide. In every case, environment config alone is not enough — hooks activate only after the code attachment shown here.

For a shorter operator/developer handoff, use the production integration playbook. It includes the attachment boundary, bypass warning and production proof checklist for every supported framework without exposing private policy or detection internals.

Runtime	SDK module	Global code location	Required code action	Tool-level enforcement

agenticdome_sdk.crewai

import agenticdome_sdk.crewai

once for global hooks, or AgenticDomeCrewAIFirewall().attach(...)

for scoped hookssecure_tool(...)

for explicit local wrapper enforcement, schema validation, sanitized-argument executionPydanticAIagenticdome_sdk.pydantic

Agent(...)

and tool is constructedCyberSecFirewall(...)

• create_hooks()

, install_native_hooks(agent)

, or attach_to_agent(agent)

@firewall.secure_tool(...)

with optional tool_schema

validationLangGraphagenticdome_sdk.langgraph

StateGraph

or LangChain create_agent()

is assembledinput_node()

/ transition_node()

/ graph_transition_node()

/ output_node()

, or as_langchain_middleware()

wrap_agent_node()

, wrap_tool_node()

, security_route()

for blocked edgesMicrosoft Agent Frameworkagenticdome_sdk.microsoft_agent_framework

create_middleware()

, install_on_agent()

, or run_agent_securely()

@firewall.secure_tool

, wrap_tool_handler

, secure_delegated_tool

, wrap_delegated_tool_handler

Microsoft AutoGenagenticdome_sdk.autogen

wrap_team()

, create_intervention_handler()

, create_termination_condition()

, attach_agentchat_agent()

, or attach_conversable_agent()

FunctionCall

authorization plus inherited wrap_tool_handler()

/ secure_tool()

Microsoft AI Foundryagenticdome_sdk.microsoft_ai_foundry

create_middleware()

, install_on_client()

, or run_secure()

wrap_tool_executor()

, @firewall.secure_tool(...)

, before_tool_call()

, delegation verifiersOpenAI Agents SDKagenticdome_sdk.openai_agents

Agent

, Runner.run(...)

, @function_tool

, guardrails, or handoffs are declaredrun_agent_securely()

, run_agent_stream_securely()

, create_input_guardrail()

, create_output_guardrail()

wrap_tool_handler()

, wrap_delegated_tool_handler()

, @firewall.secure_tool(...)

, handoff verifiersClaude Agent SDKagenticdome_sdk.claude

ClaudeAgentOptions

, ClaudeSDKClient

, SDK MCP tools, or query()

install_on_options()

plus run_client_securely()

, or secure_query()

PreToolUse

/PostToolUse

hooks, wrap_tool_handler()

, and secure_sdk_tool()

smolagentsagenticdome_sdk.smolagents

CodeAgent

, ToolCallingAgent

, tools, or managed agentsrun_agent_securely()

or attach_firewall()

Tool

wrappers, CodeAgent executor proxy, observation callback, and managed-agent token verificationAgnoagenticdome_sdk.agno

Agent

, Team, Workflow, or AgentOS components are declaredattach_firewall(agent_or_team)

, create_hook_bundle()

, create_middleware()

, or create_plugin()

@firewall.secure_tool(...)

for high-risk local toolsGoogle ADKagenticdome_sdk.google_adk

LlmAgent(...)

or ADK plugins are declaredbuild_callback_kwargs()

, create_plugin()

, or install_on_agent(...)

wrap_tool_handler()

or @firewall.secure_tool(...)

LlamaIndexagenticdome_sdk.llamaindex

run_query_securely()

, wrap_query_engine()

, wrap_retriever()

, create_node_postprocessor()

, create_callback_handler()

wrap_tool_function()

, to_function_tool()

, @firewall.secure_tool(...)

, handoff verifiersAWS Bedrockagenticdome_sdk.aws_bedrock

converse(...)

, invoke_model(...)

, invoke_agent(...)

, action-group Lambdas, or retrievalconverse_securely()

, converse_stream_securely()

, invoke_model_securely()

, invoke_model_with_response_stream_securely()

, invoke_agent_securely()

wrap_tool_handler()

, @firewall.secure_tool(...)

, wrap_action_group_lambda()

, delegation verifiersMCP host / gatewayagenticdome_sdk.mcp_host

preflight_request()

or forward_with_firewall()

around the forwarderauthorize_manager_handoff()

and verify_decision_token_if_present()

; SDK-managed security metadata never reaches the upstream serverCustom Pythonagenticdome_sdk.client

guardrail_validate()

before prompts/tools; mesh_validate()

before returning outputa2a_authorize_tool()

and a2a_verify_decision_token_rpc()

for delegationIn production, wire AgenticDome at every local boundary your process controls: prompt ingress, tool execution, delegation handoff, specialist execution, and output egress.

Every integration follows the same template: install → attach → secure tools → delegate safely → notes. Capability details and configuration snippets are collapsible so you can scan the happy path first.

One import in your bootstrap registers global hooks for before_llm_call

, before_tool_call

, and after_tool_call

— prompt screening, tool authorization, and output DLP across every crew in the process.

pip install "agenticdome-python-sdk[crewai]"
python
from crewai import Agent, Crew, Task

import agenticdome_sdk.crewai  # noqa: F401

manager = Agent(
    role="Operations Manager",
    goal="Coordinate cross-functional tasks and delegate to specialist units",
    backstory="Corporate coordinator responsible for resource routing.",
    allow_delegation=True,
)

researcher = Agent(
    role="Research Specialist",
    goal="Extract analytical records from approved secure repositories",
    backstory="Analytical expert executing restricted tasks under policy control.",
)

task = Task(
    description="Analyze database outputs and pass a summary report to the operations manager.",
    expected_output="A structured analytical report.",
    agent=manager,
)

crew = Crew(agents=[manager, researcher], tasks=[task])
result = crew.kickoff()

Scoped attachment — use the class facade for explicit hook functions, scoped attach/unregister, or a test-local client and token store (additive to the global import):

from agenticdome_sdk.crewai import AgenticDomeCrewAIFirewall

firewall = AgenticDomeCrewAIFirewall()
firewall.attach(crew)
firewall.unregister(crew)

Secure high-risk tools explicitly — if AgenticDome returns sanitized arguments, the wrapper executes the tool with those sanitized values:

@firewall.secure_tool(
    tool_name="crm.customer.read",
    tool_platform="crm",
    tool_schema={"required": ["customer_id"], "properties": {"customer_id": {"type": "string"}}},
)
def lookup_customer(agent, customer_id: str):
    return crm.get_customer(customer_id)

Security flow: (1) prompts are screened before the LLM is called; (2) tool name, clean arguments, session context, agent identity, and policy metadata are validated before execution; (3) manager→specialist delegation is authorized and can return a decision token; (4) the specialist's token is verified through the assigned runtime and consumed once using the configured SDK state store; (5) output is reviewed and can be redacted, blocked, or preserved as structured output before leaving the runtime.

CrewAI capabilities, configuration, and imports #

Supports: prompt screening before LLM calls · direct tool authorization · manager-to-specialist handoff authorization with explicit target metadata · specialist-side delegated execution verification using SDK-managed, one-time shared state · sanitized tool arguments and optional schema validation · output DLP with structured-output preservation and sanitized JSON parsing · streaming sanitization via sanitize_streaming_response()

· production mode with stable session ID requirements · local size limits, rate limits, retries/backoff, circuit breaker, audit logs, OpenTelemetry events, and emergency deny lists.

export AGENTICDOME_PLATFORM="crewai"
export AGENTICDOME_PRODUCTION_MODE="true"
export AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD="true"
export AGENTICDOME_CREWAI_MAX_INPUT_CHARS="50000"
export AGENTICDOME_CREWAI_MAX_OUTPUT_CHARS="100000"
export AGENTICDOME_CREWAI_MAX_TOOL_ARG_CHARS="20000"
export AGENTICDOME_CREWAI_RATE_LIMIT_PER_MINUTE="120"
export AGENTICDOME_CREWAI_RETRY_ATTEMPTS="2"
export AGENTICDOME_CREWAI_CIRCUIT_BREAKER_FAILURES="5"
export AGENTICDOME_CREWAI_AUDIT_LOGGING="true"
export AGENTICDOME_CREWAI_OTEL_ENABLED="true"
python
import agenticdome_sdk.crewai

from agenticdome_sdk.crewai import (
    CONFIG,
    CLIENT,
    AgenticDomeCrewAIFirewall,
    DecisionTokenRecord,
    DecisionTokenStore,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
    AgenticDome_before_tool_call,
    AgenticDome_after_tool_call,
    AgenticDome_before_llm_call,
    sanitize_streaming_response,
    attach_firewall,
    unregister_firewall,
)

Attach lifecycle hooks where each Agent(...)

is constructed, and always decorate tools that access data, systems, or external actions.

pip install "agenticdome-python-sdk[pydanticai]"
python
import os
from typing import Any

from pydantic_ai import Agent, RunContext
from agenticdome_sdk.pydantic import CyberSecFirewall, FirewallConfig

firewall = CyberSecFirewall(
    config=FirewallConfig(
        api_base=os.environ["AGENTICDOME_API_BASE"],
        api_key=os.environ["AGENTICDOME_API_KEY"],
        tenant_id=os.environ["AGENTICDOME_TENANT_ID"],
        fail_closed=True,
        block_on_sensitive_output=True,
    )
)

customer_support_agent = Agent(
    "gemini-2.5-flash",
    name="customer_support_agent",
    result_type=str,
    system_prompt="You are a helpful customer platform support assistant.",
)

firewall.install_native_hooks(customer_support_agent)

firewall.attach_to_agent(customer_support_agent)

@customer_support_agent.tool
@firewall.secure_tool(
    tool_name="customer.profile.read",
    tool_platform="crm",
    tool_schema={
        "required": ["user_id"],
        "properties": {"user_id": {"type": "string"}},
    },
)
async def fetch_user_profile(ctx: RunContext[Any], user_id: str) -> dict:
    """Retrieves account management metadata profiles for a corporate ID."""
    return {
        "user_id": user_id,
        "status": "active",
        "passport_number": "A-1234567",
    }

Manual firewall usage — in custom routers, test harnesses, or execution gateways:

async for safe_chunk in firewall.sanitize_streaming_response(
    chunks=agent_stream,
    agent_id="customer_support_agent",
    session_id="sess_prod_01J4X",
):
    yield safe_chunk

PydanticAI capabilities, version notes, and imports #

Supports: prompt ingress checks via legacy lifecycle hooks where available · native Hooks

capability creation through create_hooks()

for current PydanticAI versions · tool perimeter authorization via @firewall.secure_tool(...)

· Pydantic/JSON-schema argument validation and sanitized-argument execution · manager/specialist delegation with SDK-managed, integrity-protected shared state · egress output DLP with correct block_on_sensitive_output

semantics · structured-output preservation (sanitized JSON parsed back to dicts/lists) · stable session ID enforcement in production mode · local rate limits, size limits, retries, circuit breaker, audit logging, OpenTelemetry span events · streaming sanitization · identity-rich policy context from ctx

, deps

, or nested identity/principal objects · emergency deny lists.

Version notes: PydanticAI lifecycle hook APIs have evolved. Current PydanticAI documents pydantic_ai.capabilities.Hooks

for lifecycle interception across runs, model requests, tool validation/execution, output processing, and event streams. Prefer create_hooks()

/ install_native_hooks()

on current runtimes; keep @firewall.secure_tool(...)

on sensitive tools as a hard enforcement boundary. If legacy lifecycle decorators are available, attach_to_agent()

attaches prompt ingress and egress DLP hooks; if not, @firewall.secure_tool(...)

still protects tool execution. AGENTICDOME_BLOCK_ON_SENSITIVE_OUTPUT=true

means AgenticDome may ask Mesh to block sensitive output; the SDK only blocks when the policy response verdict is BLOCKED

.

from agenticdome_sdk.pydantic import (
    CyberSecFirewall,
    FirewallConfig,
    PydanticAIFirewallError,
    PydanticAIFirewallDenied,
    PydanticAIFirewallConfigurationError,
    DecisionTokenRecord,
    DecisionTokenStore,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
)

Three production patterns: explicit firewall nodes when you own the graph topology, wrappers when you already have nodes, and LangChain middleware when you use create_agent(..., middleware=[...])

.

pip install "agenticdome-python-sdk[langgraph]"

Pattern 1 — explicit firewall nodes (clear security boundaries before input, before tools, before final output):

import os
from langgraph.graph import END, START, StateGraph
from agenticdome_sdk.langgraph import AgentState, AgenticDomeLangGraphFirewall, FirewallConfig

firewall = AgenticDomeLangGraphFirewall(
    config=FirewallConfig(
        api_base=os.environ["AGENTICDOME_API_BASE"],
        api_key=os.environ["AGENTICDOME_API_KEY"],
        tenant_id=os.environ["AGENTICDOME_TENANT_ID"],
        fail_closed=True,
        production_mode=True,
        require_explicit_session_id=True,
        rate_limit_per_minute=120,
        max_tool_arg_chars=20_000,
    )
)

async def agent_node(state: AgentState) -> AgentState:
    return state

async def tool_node(state: AgentState) -> AgentState:
    return state

graph = StateGraph(AgentState)
graph.add_node("input_firewall", firewall.input_node(agent_id="support_orchestrator"))
graph.add_node("agent", agent_node)
graph.add_node("transition_firewall", firewall.transition_node(agent_id="support_orchestrator"))
graph.add_node("tools", tool_node)
graph.add_node("output_firewall", firewall.output_node(agent_id="support_orchestrator"))

graph.add_edge(START, "input_firewall")
graph.add_edge("input_firewall", "agent")
graph.add_edge("agent", "transition_firewall")
graph.add_edge("transition_firewall", "tools")
graph.add_edge("tools", "output_firewall")
graph.add_edge("output_firewall", END)

compiled = graph.compile()

result = await compiled.ainvoke({
    "session_id": "sess_prod_01J4X",
    "messages": [{"role": "user", "content": "Check the customer refund status."}],
    "agent_id": "support_orchestrator",
})

Pattern 2 — wrap existing nodes without changing their internals:

secure_agent_node = firewall.wrap_agent_node(
    existing_agent_node,
    agent_id="claims_agent",
    screen_input=True,
    sanitize_output=True,
)

secure_tool_node = firewall.wrap_tool_node(
    existing_tool_node,
    agent_id="claims_agent",
    sanitize_tool_output=True,
)

Pattern 3 — LangChain agent middleware:

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-4.1-mini",
    tools=[lookup_customer, create_refund],
    middleware=[firewall.as_langchain_middleware(agent_id="support_agent")],
)

Delegation — a manager node requests a specialist handoff by adding an AgenticDome.handoff

(or top-level handoff

) payload to graph state; the firewall authorizes it and stores the decision token in state plus the token store:

state["AgenticDome"] = {
    "handoff": {
        "target_agent_id": "refund_specialist",
        "delegated_tool_name": "payments.refund.create",
        "delegated_tool_args": {
            "customer_id": "cust_123",
            "amount": 250,
            "currency": "AUD",
        },
        "tool_platform": "payments",
        "text": "Manager delegates refund execution to a specialist agent.",
    }
}

Specialist execution is verified when the delegated tool call reaches authorize_transition()

or a wrapped tool node. The SDK carries or recovers integrity-protected delegation state and consumes it once. Applications should use the public handoff/wrapper APIs rather than create, inspect or forward the SDK's internal metadata.

Hardening helpers — policy-control sensitive graph edges; blocked states set AgenticDome.route

and next_agent_id

to security_block

:

graph.add_node(
    "authorize_escalation",
    firewall.graph_transition_node(
        from_node="triage",
        to_node="refund_specialist",
        agent_id="support_orchestrator",
    ),
)

graph.add_conditional_edges(
    "authorize_escalation",
    firewall.security_route,
    {"continue": "refund_specialist", "security_block": "security_block"},
)

Use sanitize_retrieval_documents()

before adding retrieved chunks to model context, and sanitize_streaming_events()

for async event streams — both use the same Mesh output policy path as final-output sanitization.

LangGraph capabilities, interception notes, and imports #

Supports: prompt ingress via screen_input()

/ input_node()

· tool-call authorization via authorize_transition()

/ transition_node()

· delegation authorization from documented handoff fields · specialist-side verification using SDK-managed one-time state · sanitized tool-argument mutation before local execution · final message and tool-output DLP via sanitize_output()

/ output_node()

· retrieval and streaming sanitization · graph transition authorization · security_block

routing · wrappers for existing agent and tool nodes.

Interception notes: LangGraph is graph-native — reliable interception means inserting security nodes or wrapping nodes/tool nodes. LangChain's modern create_agent()

supports a middleware

parameter documented as the way to intercept model, tool, and agent-loop behavior; as_langchain_middleware()

targets that style, and the adapter mutates sanitized tool arguments back into the tool request for local execution. For custom StateGraph

workflows, implement middleware as graph nodes or wrappers at the boundaries you must enforce: before model input, before tool execution, before handoff execution, and before final output. Remote or provider-hosted tools that execute outside your Python process can only be guarded at the local request/response boundary.

Official references: LangChain middleware overview · LangChain create_agent reference

from agenticdome_sdk.langgraph import (
    AgentState,
    AgenticDomeLangGraphFirewall,
    AgenticDomeLangChainMiddleware,
    FirewallConfig,
    AgenticDomeDenied,
    AgenticDomeConfigurationError,
    DecisionTokenRecord,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
)

Boundary-oriented async firewall: protect the run boundary, local function-tool handlers, delegated specialist tools, and final output. It does not monkey-patch every Microsoft provider or hosted tool surface.

pip install "agenticdome-python-sdk[microsoft]"

Install the Microsoft Agent Framework packages used by your application separately — the AgenticDome helper is dependency-light because deployments vary across local function tools, hosted tools, Foundry agents, Copilot Studio, A2A agents, workflow executors, and custom clients.

Secure a local function tool — wrap the callable that actually executes, so arguments are authorized before execution and results sanitized after. If AgenticDome returns sanitized_tool_args

, the wrapped handler receives those safe arguments instead of the model-provided originals:

import os
from typing import Annotated

from pydantic import Field
from agent_framework import tool
from agenticdome_sdk.microsoft_agent_framework import AgenticDomeMicrosoftAgentFirewall

firewall = AgenticDomeMicrosoftAgentFirewall()

async def raw_get_customer_profile(ctx, args):
    customer_id = args["customer_id"]
    return {
        "customer_id": customer_id,
        "email": "alice@example.com",
        "risk": "medium",
    }

secure_get_customer_profile = firewall.wrap_tool_handler(
    tool_name="crm.customer_profile.read",
    handler=raw_get_customer_profile,
    tool_platform="crm",
)

@tool(approval_mode="never_require")
async def get_customer_profile(
    customer_id: Annotated[str, Field(description="Customer identifier")],
) -> str:
    ctx = {
        "session_id": "sess_prod_01J4X",
        "agent_name": "customer_support_agent",
    }
    return await secure_get_customer_profile(ctx, {"customer_id": customer_id})

Native-style middleware hooks — a harder-to-bypass assembly-level integration where the runtime exposes middleware or callbacks; before_tool_call()

returns the sanitized tool arguments to forward to the local executor:

firewall = AgenticDomeMicrosoftAgentFirewall()

agent = firewall.install_on_agent(agent)
middleware = firewall.create_middleware()

Secure the whole agent run boundary — prompt ingress plus final-output DLP:

result = await firewall.run_agent_securely(
    run_callable=agent.run,
    input_text="Find the customer's refund status.",
    session_id="sess_prod_01J4X",
    agent_id="refund_agent",
    policy_context={"request_purpose": "customer_support"},
    output_extractor=lambda value: getattr(value, "text", str(value)),
)

Delegated specialist pattern — authorize at the manager, verify at the specialist:

authorization = await firewall.authorize_manager_handoff(
    text="Manager delegates refund execution to a payment specialist.",
    manager_agent_id="support_manager",
    specialist_agent_id="payments_specialist",
    tool_name="payments.refund.create",
    tool_args={"customer_id": "cust_123", "amount": 250, "currency": "AUD"},
    session_id="sess_prod_01J4X",
    tool_platform="payments",
)

secure_refund_handler = firewall.wrap_delegated_tool_handler(
    tool_name="payments.refund.create",
    handler=raw_refund_handler,
)

Microsoft Agent Framework capabilities, configuration, notes, and imports #

Supports: prompt ingress via screen_input()

, middleware hooks, or run_agent_securely()

· function-tool authorization with sanitized-argument enforcement · manager-to-specialist delegation and specialist verification through public wrapper APIs · stable session ID enforcement for production · Entra/principal identity context propagation · output DLP with structured JSON preservation and optional response-object mutation · streaming sanitization · OpenTelemetry events and structured audit logging · local rate limits, size limits, retries, circuit breaker · optional Copilot / AI Foundry threat helper enforcement · shared multi-worker delegation state · emergency deny lists.

export AGENTICDOME_PLATFORM="microsoft_agent_framework_v1"
export AGENTICDOME_PRODUCTION_MODE="true"
export AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD="true"
export AGENTICDOME_MSAF_MAX_INPUT_CHARS="50000"
export AGENTICDOME_MSAF_MAX_OUTPUT_CHARS="100000"
export AGENTICDOME_MSAF_MAX_TOOL_ARG_CHARS="20000"
export AGENTICDOME_MSAF_RATE_LIMIT_PER_MINUTE="0"
export AGENTICDOME_MSAF_RETRY_ATTEMPTS="2"
export AGENTICDOME_MSAF_CIRCUIT_BREAKER_FAILURES="5"
export AGENTICDOME_MSAF_CIRCUIT_BREAKER_RESET_S="60"
export AGENTICDOME_MSAF_AUDIT_LOGGING="true"
export AGENTICDOME_MSAF_OTEL_ENABLED="true"

Notes: the framework's tool-approval feature is human-in-the-loop gating, not policy enforcement, DLP, or tenant-aware A2A token verification — AgenticDome should sit at the local tool handler or workflow executor boundary for deterministic enforcement. Wrap the executor/run boundary or the tool/executor functions that process sensitive actions. Tools that execute remotely (hosted providers, Foundry agents, Copilot Studio, hosted MCP servers, remote A2A agents) can only be protected at the local request/response boundary. In production, pass stable session_id

/run_id

/trace_id

values and Entra/principal identity fields in context.

Official references: Agent Framework docs · Tools overview · Workflow execution

from agenticdome_sdk.microsoft_agent_framework import (
    AgenticDomeMicrosoftAgentFirewall,
    FirewallConfig,
    load_config,
    MicrosoftAgentFirewallDenied,
    MicrosoftAgentFirewallError,
    DecisionTokenRecord,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
)

AutoGen is Microsoft's open-source conversational multi-agent framework and is now community-managed in maintenance mode; Microsoft Agent Framework is the recommended successor for new systems. AgenticDome supports current AutoGen AgentChat/Core applications and existing legacy ConversableAgent

deployments so teams can migrate without losing runtime enforcement.

pip install "agenticdome-python-sdk[autogen]"

Protect a current AgentChat team — the wrapper screens the initial task, every streamed team event, and final messages while retaining the underlying Team API:

from autogen_agentchat.teams import RoundRobinGroupChat
from agenticdome_sdk.autogen import AgenticDomeAutoGenFirewall

firewall = AgenticDomeAutoGenFirewall()
team = RoundRobinGroupChat([planner, researcher, payments_specialist], max_turns=12)
secure_team = firewall.wrap_team(
    team,
    session_id="sess_prod_01J4X",
    agent_id="customer_operations_team",
    policy_context={"request_purpose": "customer_support"},
)

result = await secure_team.run(task=user_prompt)

Authorize AutoGen Core tool traffic at the runtime boundary — current AutoGen sends FunctionCall

messages to tool agents, so the intervention handler is a stronger boundary than patching an individual assistant:

from autogen_core import SingleThreadedAgentRuntime

handler = firewall.create_intervention_handler(
    session_id="sess_prod_01J4X",
    agent_id="autogen_planner",
)
runtime = SingleThreadedAgentRuntime(intervention_handlers=[handler])

Freeze a group chat on behavioral drift — compose the AgenticDome condition with AutoGen's normal termination conditions:

agenticdome_stop = firewall.create_termination_condition(
    session_id="sess_prod_01J4X",
    agent_id="customer_operations_team",
)
team = RoundRobinGroupChat(
    [planner, researcher, payments_specialist],
    termination_condition=agenticdome_stop | normal_stop,
    max_turns=12,
)

Family 2 policy receives a bounded rolling conversation window digest, participant lineage, semantic-deviation evaluation request, and tool-call frequency. A blocked cross-agent message or excessive tool rate freezes the local session, reports a trust incident, and advances revocation state for the emitting agent before an external action can run.

Existing AutoGen 0.2 deployments — attach to the legacy ConversableAgent.send()

, receive()

, a_send()

, and a_receive()

lifecycles:

Keep the customer's already-certified legacy AutoGen dependency in place and install the dependency-light base SDK (do not use the [autogen]

extra, because that extra deliberately installs the current AgentChat release):

pip install agenticdome-python-sdk
assistant = firewall.attach_conversable_agent(
    assistant,
    session_id="sess_prod_01J4X",
    agent_id="legacy_autogen_assistant",
)
user_proxy = firewall.attach_conversable_agent(
    user_proxy,
    session_id="sess_prod_01J4X",
    agent_id="legacy_autogen_user_proxy",
)

Wrap side-effecting local tools as well; conversation screening does not replace authorization at the execution boundary:

secure_refund = firewall.wrap_tool_handler(
    tool_name="payments.refund.create",
    tool_platform="payments",
    handler=raw_refund,
    session_id="sess_prod_01J4X",
    agent_id="payments_specialist",
)
export AGENTICDOME_PLATFORM="autogen"
export AGENTICDOME_PRODUCTION_MODE="true"
export AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD="true"
export AGENTICDOME_AUTOGEN_CONVERSATION_WINDOW="12"
export AGENTICDOME_AUTOGEN_MAX_TOOL_CALLS_PER_WINDOW="8"
export AGENTICDOME_AUTOGEN_FREEZE_ON_BLOCK="true"
export AGENTICDOME_AUTOGEN_REVOKE_ON_FREEZE="true"

Official references: AutoGen project status and migration guidance · AgentChat teams · Core intervention handlers · Legacy 0.2 conversational agents

For services that call Foundry agents, handle function-call requests from Foundry, execute local function tools, or use FoundryChatClient

with local tools.

pip install "agenticdome-python-sdk[foundry]"

The adapter itself is dependency-light; the [foundry]

extra installs common Azure SDK packages (azure-ai-projects

, azure-identity

) for applications using Foundry directly.

Authentication model — Foundry threat-contract calls use bearer auth; Mesh output DLP and incident reporting use API-key auth. In production mode, output sanitization is required by default:

export AGENTICDOME_API_BASE="https://demo-sidecar.agenticdome.io"
export AGENTICDOME_BEARER_TOKEN="your_foundry_threat_contract_bearer_token"

export AGENTICDOME_API_KEY="your_api_key"
export AGENTICDOME_TENANT_ID="your_tenant_id"
export AGENTICDOME_PRODUCTION_MODE="true"
export AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD="true"
export AGENTICDOME_FOUNDRY_REQUIRE_OUTPUT_SANITIZATION_IN_PROD="true"

Attach middleware, then secure the run boundary:

from agenticdome_sdk.microsoft_ai_foundry import AgenticDomeMicrosoftAIFoundryFirewall

firewall = AgenticDomeMicrosoftAIFoundryFirewall()
foundry_client = firewall.install_on_client(foundry_client)

middleware = firewall.create_middleware()
await middleware.before_run(ctx, input_text)
result = await foundry_agent.run(input_text)
result = await middleware.after_run(ctx, result)
result = await firewall.run_secure(
    run_callable=foundry_agent.run,
    input_text="Find the customer's refund status.",
    ctx={
        "agent_id": "foundry_refund_agent",
        "session_id": "sess_prod_01J4X",
        "user_id": "user_123",
    },
    output_extractor=lambda value: getattr(value, "text", str(value)),
)

Secure local function-tool execution — at the exact boundary before your app submits function output back to Foundry:

async def raw_lookup_customer(ctx, args):
    return {"customer_id": args["customer_id"], "email": "alice@example.com"}

secure_lookup_customer = firewall.wrap_tool_executor(
    tool_name="crm.customer.read",
    tool_platform="crm",
    handler=raw_lookup_customer,
    tool_schema={
        "required": ["customer_id"],
        "properties": {"customer_id": {"type": "string"}},
    },
)

result = await secure_lookup_customer(
    {"agent_id": "foundry_support_agent", "session_id": "sess_prod_01J4X"},
    {"customer_id": "cust_123"},
)

Decorator form:

@firewall.secure_tool(
    tool_name="payments.refund.create",
    tool_platform="payments",
    tool_schema={
        "required": ["customer_id", "amount_cents"],
        "properties": {
            "customer_id": {"type": "string"},
            "amount_cents": {"type": "integer"},
        },
    },
)
def create_refund(ctx, args):
    return {"refund_id": "rfnd_123", "status": "created"}

Delegated Foundry tool execution — authorization stores decision state in memory by default. Configure the optional Redis store only when the specialist executes in another process, worker, or pod; the specialist consumes and verifies the decision before executing:

await firewall.authorize_manager_handoff(
    text="Ask the billing specialist to create a refund.",
    manager_agent_id="foundry_manager",
    specialist_agent_id="billing_specialist",
    tool_name="payments.refund.create",
    tool_args={"customer_id": "cust_123", "amount_cents": 2500},
    session_id="sess_prod_01J4X",
    tool_platform="payments",
)

await firewall.verify_delegated_execution(
    specialist_agent_id="billing_specialist",
    tool_name="payments.refund.create",
    tool_args={"customer_id": "cust_123", "amount_cents": 2500},
    session_id="sess_prod_01J4X",
)

Streaming output sanitization:

async for safe_chunk in firewall.sanitize_streaming_response(
    chunks=foundry_stream,
    agent_id="foundry_support_agent",
    session_id="sess_prod_01J4X",
):
    yield safe_chunk

Microsoft AI Foundry capabilities, notes, and imports #

Supports: prompt/run validation via validate_prompt_contract()

, before_run()

, or run_secure()

· middleware hooks via create_middleware()

/ install_on_client()

· local function-tool analysis via analyze_tool_execution()

, before_tool_call()

, or wrap_tool_executor()

· @firewall.secure_tool(...)

for high-risk callables · lightweight JSON-schema validation and sanitized-argument execution · enterprise identity context propagation for Entra IDs, roles/scopes, Foundry project IDs, and Purview/sensitivity labels · production-mode stable session ID and output-sanitization requirements · output DLP through Mesh · structured-output preservation · local rate limits, size limits, retries, circuit breaker, audit logging, OpenTelemetry span events · streaming sanitization · optional handoff authorization and SDK-managed multi-worker verification · emergency deny lists.

Notes: Foundry function calling asks your application to execute local functions and return tool output — wrap that local execution before output is submitted back to Foundry. Production deployments should pass a stable session_id

, run_id

, trace_id

, conversation_id

, or thread_id

; generated fallback IDs are for local development only. Pass Entra identity, roles/scopes, Foundry project IDs, and Purview/sensitivity labels on ctx

or policy_context

for identity-aware server-side policy. Hosted tools executing entirely inside a remote provider runtime can only be protected at the local request/response boundary. Threat-contract prompt and tool analysis additionally require AGENTICDOME_BEARER_TOKEN

.

Official references: Foundry function calling · Foundry agents quickstart

from agenticdome_sdk.microsoft_ai_foundry import (
    AgenticDomeMicrosoftAIFoundryFirewall,
    FirewallConfig,
    MicrosoftAIFoundryDenied,
    MicrosoftAIFoundryFirewallError,
    MicrosoftAIFoundryConfigurationError,
    DecisionTokenRecord,
    DecisionTokenStore,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
)

The OpenAI Agents SDK ships agents, function tools, guardrails, handoffs, sessions, streaming, and tracing; AgenticDome complements those primitives by enforcing tenant policy before local tool execution, validating delegated specialist execution, and sanitizing outputs before they leave the runtime.

pip install "agenticdome-python-sdk[openai-agents]"   # installs the openai-agents package

Secure a runner boundary:

from agents import Agent, Runner
from agenticdome_sdk.openai_agents import AgenticDomeOpenAIAgentsFirewall

firewall = AgenticDomeOpenAIAgentsFirewall()
agent = Agent(name="support_agent", instructions="Help support users safely.")

result = await firewall.run_agent_securely(
    runner=Runner,
    agent=agent,
    input_text="Check customer refund status.",
    session_id="sess_prod_01J4X",
)

For streamed runs, use run_agent_stream_securely()

or pass the stream through sanitize_streaming_response()

before returning chunks.

Register guardrail helpers where your wiring supports input/output guardrail slots — but keep tool authorization at function-tool boundaries, because tool execution can happen multiple times inside one run:

input_guardrail = firewall.create_input_guardrail()
output_guardrail = firewall.create_output_guardrail()

Secure a function tool — wrap the local implementation before exposing it with @function_tool

; sanitized arguments replace originals and SDK-managed security metadata is never passed to the business handler:

from agents import function_tool

async def raw_lookup_customer(ctx, args):
    return {"customer_id": args["customer_id"], "email": "alice@example.com"}

secure_lookup_customer = firewall.wrap_tool_handler(
    tool_name="crm.customer.read",
    tool_platform="crm",
    tool_schema={"required": ["customer_id"], "properties": {"customer_id": {"type": "string"}}},
    handler=raw_lookup_customer,
)

@function_tool
async def lookup_customer(customer_id: str) -> str:
    return await secure_lookup_customer(
        {"agent_id": "support_agent", "session_id": "sess_prod_01J4X"},
        {"customer_id": customer_id},
    )

Delegated specialist tool pattern:

await firewall.authorize_manager_handoff(
    session_id="sess_prod_01J4X",
    manager_agent_id="triage_agent",
    specialist_agent_id="refund_agent",
    tool_name="payments.refund.create",
    tool_args={"customer_id": "cust_123", "amount": 250},
    text="Triage agent delegates refund creation to refund specialist.",
    tool_platform="payments",
)

secure_refund_tool = firewall.wrap_delegated_tool_handler(
    tool_name="payments.refund.create",
    handler=raw_refund_handler,
)

OpenAI Agents SDK capabilities, configuration, notes, and imports #

Supports: prompt ingress via screen_input()

, run_agent_securely()

, run_agent_stream_securely()

, or create_input_guardrail()

· function-tool authorization via wrap_tool_handler()

/ @firewall.secure_tool(...)

· sanitized arguments and optional schema validation · handoff authorization via authorize_manager_handoff()

· specialist-side verification via verify_specialist_execution()

and wrap_delegated_tool_handler()

· SDK-managed one-time multi-worker delegation state · output DLP via sanitize_output()

and create_output_guardrail()

· streaming sanitization · structured-output preservation and sanitized JSON parsing · production mode with stable session IDs · size limits, rate limits, retries/backoff, circuit breaker, audit logs, OpenTelemetry events, identity-rich policy context, emergency deny lists.

export AGENTICDOME_PLATFORM="openai_agents_sdk"
export AGENTICDOME_PRODUCTION_MODE="true"
export AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD="true"
export AGENTICDOME_OPENAI_AGENTS_MAX_INPUT_CHARS="50000"
export AGENTICDOME_OPENAI_AGENTS_MAX_OUTPUT_CHARS="100000"
export AGENTICDOME_OPENAI_AGENTS_MAX_TOOL_ARG_CHARS="20000"
export AGENTICDOME_OPENAI_AGENTS_RATE_LIMIT_PER_MINUTE="120"
export AGENTICDOME_OPENAI_AGENTS_RETRY_ATTEMPTS="2"
export AGENTICDOME_OPENAI_AGENTS_CIRCUIT_BREAKER_FAILURES="5"
export AGENTICDOME_OPENAI_AGENTS_AUDIT_LOGGING="true"
export AGENTICDOME_OPENAI_AGENTS_OTEL_ENABLED="true"

Notes: guardrails are useful at run boundaries, but side-effecting local tools still need function-tool wrappers. Handoffs are represented as tools to the model, so manager-to-specialist policy should be enforced where handoff/tool execution is invoked. Hosted tools, MCP tools, and remote runtimes can only be protected at the local request/response boundary. Use stable session_id

/run_id

/trace_id

/conversation_id

/thread_id

values and Redis-backed token storage when authorization and execution can happen in different workers.

Official references: Overview · Tools · Guardrails · Handoffs

from agenticdome_sdk.openai_agents import (
    AgenticDomeOpenAIAgentsFirewall,
    FirewallConfig,
    OpenAIAgentsFirewallDenied,
    OpenAIAgentsFirewallError,
    DecisionTokenRecord,
    DecisionTokenStore,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
)

The adapter uses Claude Agent SDK's native hook contract for prompt submission, pre-tool permission decisions, and post-tool output replacement. It also wraps the asynchronous query()

and ClaudeSDKClient.receive_response()

pipelines so final assistant text is reviewed before your application returns it.

pip install "agenticdome-python-sdk[claude]"

Secure a ClaudeSDKClient and its built-in/MCP tools:

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
from agenticdome_sdk.claude import AgenticDomeClaudeFirewall

firewall = AgenticDomeClaudeFirewall()
options = ClaudeAgentOptions(allowed_tools=["Read", "mcp__crm__lookup"])
firewall.install_on_options(
    options,
    session_id="sess_prod_01J4X",
    agent_id="claude_support_agent",
)

async with ClaudeSDKClient(options=options) as client:
    async for message in firewall.run_client_securely(
        client,
        "Look up the customer's active support case.",
        session_id="sess_prod_01J4X",
        agent_id="claude_support_agent",
    ):
        consume(message)

For the one-shot API, iterate firewall.secure_query(prompt, session_id=..., options=...)

. If the run may execute built-in tools, install the returned hook matchers on its options as well; secure_query()

itself covers ingress and returned messages.

Compose with Claude's native SDK MCP @tool:

@firewall.secure_sdk_tool(
    "lookup_customer",
    "Look up a customer support profile",
    {"customer_id": str},
    session_id="sess_prod_01J4X",
    agent_id="claude_support_agent",
    tool_platform="crm",
)
async def lookup_customer(args):
    return {"content": [{"type": "text", "text": crm_lookup(args["customer_id"])}]}

The PreToolUse

hook returns Claude's native permissionDecision: deny

response before local side effects. If policy supplies sanitized arguments, it returns updatedInput

. The PostToolUse

hook uses updatedToolOutput

so DLP-reviewed tool data is what the model sees.

export AGENTICDOME_PLATFORM="claude_agent_sdk"
export AGENTICDOME_CLAUDE_AGENT_ID="claude_support_agent"
export AGENTICDOME_CLAUDE_MAX_INPUT_CHARS="50000"
export AGENTICDOME_CLAUDE_MAX_OUTPUT_CHARS="100000"
export AGENTICDOME_CLAUDE_MAX_TOOL_ARG_CHARS="20000"
export AGENTICDOME_CLAUDE_STREAMING_BUFFER_CHARS="4000"
export AGENTICDOME_CLAUDE_RATE_LIMIT_PER_MINUTE="60"
export AGENTICDOME_CLAUDE_RETRY_ATTEMPTS="2"
export AGENTICDOME_CLAUDE_RETRY_BACKOFF_S="0.25"
export AGENTICDOME_CLAUDE_CIRCUIT_BREAKER_FAILURES="5"
export AGENTICDOME_CLAUDE_CIRCUIT_BREAKER_RESET_S="60"
export AGENTICDOME_CLAUDE_AUDIT_LOGGING="true"
export AGENTICDOME_CLAUDE_OTEL_ENABLED="true"
export AGENTICDOME_CLAUDE_STRICT_DELEGATED_EXECUTION="true"
export AGENTICDOME_CLAUDE_EMERGENCY_BLOCK_TOOLS=""
export AGENTICDOME_CLAUDE_EMERGENCY_BLOCK_AGENTS=""

Use authorize_manager_handoff()

and verify_specialist_execution()

when a manager delegates sensitive work. Configure the documented shared store and integrity secret when authorization and specialist execution can land on different workers. Claude hooks protect operations visible to the local SDK process; externally hosted services still require enforcement at their local gateway or MCP host.

Official references: Claude Agent SDK Python · Claude Agent SDK overview

smolagents CodeAgent

generates Python and invokes python_executor(code)

before step callbacks run. The adapter therefore wraps the executor itself, wraps every native Tool

, sanitizes step observations before the next model turn, and enforces managed-agent handoffs with bound decision tokens.

pip install "agenticdome-python-sdk[smolagents]"
python
from smolagents import CodeAgent, InferenceClientModel, tool
from agenticdome_sdk.smolagents import AgenticDomeSmolagentsFirewall

@tool
def lookup_customer(customer_id: str) -> str:
    """Look up a customer by ID."""
    return crm_lookup(customer_id)

agent = CodeAgent(tools=[lookup_customer], model=InferenceClientModel())
firewall = AgenticDomeSmolagentsFirewall()

result = firewall.run_agent_securely(
    agent,
    "Look up customer cust_123 for their active support case.",
    session_id="sess_prod_01J4X",
    agent_id="smol_support_agent",
)

attach_firewall(agent, session_id=...)

is idempotent and can be used when another component owns agent.run()

. For streaming, use run_agent_stream_securely()

so event output is reviewed before it is yielded. Direct agent.run()

after attachment still gets tool, code, managed-agent, and step-observation enforcement, but the application should use the secure run wrapper for final-output DLP.

export AGENTICDOME_PLATFORM="smolagents"
export AGENTICDOME_SMOLAGENTS_AGENT_ID="smol_support_agent"
export AGENTICDOME_SMOLAGENTS_MAX_INPUT_CHARS="50000"
export AGENTICDOME_SMOLAGENTS_MAX_OUTPUT_CHARS="100000"
export AGENTICDOME_SMOLAGENTS_MAX_TOOL_ARG_CHARS="20000"
export AGENTICDOME_SMOLAGENTS_STREAMING_BUFFER_CHARS="4000"
export AGENTICDOME_SMOLAGENTS_RATE_LIMIT_PER_MINUTE="60"
export AGENTICDOME_SMOLAGENTS_RETRY_ATTEMPTS="2"
export AGENTICDOME_SMOLAGENTS_RETRY_BACKOFF_S="0.25"
export AGENTICDOME_SMOLAGENTS_CIRCUIT_BREAKER_FAILURES="5"
export AGENTICDOME_SMOLAGENTS_CIRCUIT_BREAKER_RESET_S="60"
export AGENTICDOME_SMOLAGENTS_AUDIT_LOGGING="true"
export AGENTICDOME_SMOLAGENTS_OTEL_ENABLED="true"
export AGENTICDOME_SMOLAGENTS_EMERGENCY_BLOCK_TOOLS=""
export AGENTICDOME_SMOLAGENTS_EMERGENCY_BLOCK_AGENTS=""
export AGENTICDOME_SMOLAGENTS_STRICT_DELEGATED_EXECUTION="true"
export AGENTICDOME_SMOLAGENTS_SCAN_CODE_EXPRESSIONS="true"

Keep code-expression scanning enabled in production. It adds business-intent policy before smolagents' local or remote executor; it does not replace the executor's OS/container/WASM sandbox. The adapter intentionally sends generated code and serialized tool arguments to the configured AgenticDome sidecar, so place that sidecar within the approved trust boundary and apply normal data-residency controls.

Official references: smolagents agents · smolagents tools

Agno's Agent reference documents pre_hooks

, post_hooks

, and tool_hooks

; the adapter attaches to those boundaries so policy is enforced before prompts/tools run and before output returns. Middleware/plugin-shaped helpers are available for applications that centralize hook registration.

pip install "agenticdome-python-sdk[agno]"     # install agno separately as needed

Attach firewall hooks in the module where you create the Agno Agent

, Team, Workflow, or AgentOS component (attach_firewall()

is idempotent):

from agno.agent import Agent
from agenticdome_sdk.agno import AgenticDomeAgnoFirewall

firewall = AgenticDomeAgnoFirewall()

support_agent = Agent(
    name="support_agent",
    model="openai:gpt-5.5",
    tools=[lookup_customer, create_refund],
)

firewall.attach_firewall(support_agent)
php
pre_hooks   -> prompt input, tool authorization, delegation authorization, token verification
post_hooks  -> final output DLP and redaction/blocking
tool_hooks  -> additional local tool boundary enforcement where Agno invokes tool hooks

For centralized registration layers:

hook_bundle = firewall.create_hook_bundle()
middleware = firewall.create_middleware()
plugin = firewall.create_plugin()

Decorate high-risk tools — anything that reads sensitive data, mutates state, sends messages, writes files, calls payment systems, or triggers external APIs:

@firewall.secure_tool(
    tool_name="crm.customer.read",
    tool_platform="crm",
    tool_schema={"required": ["customer_id"], "properties": {"customer_id": {"type": "string"}}},
)
def lookup_customer(agent, customer_id: str) -> dict:
    return {"customer_id": customer_id, "email": "alice@example.com"}

Delegation — pass target metadata in hook kwargs or tool args; AgenticDome authorizes the handoff and stores the decision token for specialist verification:

firewall.pre_hook(
    manager_agent,
    session_id="sess_prod_01J4X",
    input="Delegate refund execution to payment specialist.",
    tool_name="delegate_refund",
    tool_platform="payments",
    tool_args={
        "target_agent_id": "payments_specialist",
        "target_tool_name": "payments.refund.create",
        "target_tool_args": {"customer_id": "cust_123", "amount": 250},
    },
)

The specialist side verifies a token passed in args or recovers it from the configured in-process or optional Redis store; stored tokens are consumed once:

firewall.pre_hook(
    payments_specialist,
    session_id="sess_prod_01J4X",
    tool_name="payments.refund.create",
    tool_args={"customer_id": "cust_123", "amount": 250},
)

Retrieved context and streaming sanitization — before retrieved or streamed content is shown to a user or re-enters an agent loop:

safe_context = firewall.sanitize_retrieved_text(
    text=retrieved_context,
    agent_id="support_agent",
    session_id="sess_prod_01J4X",
    policy_context={"source": "agno_knowledge"},
)

async for safe_chunk in firewall.sanitize_streaming_response(
    chunks,
    agent_id="support_agent",
    session_id="sess_prod_01J4X",
):
    yield safe_chunk

Agno capabilities, configuration, notes, and imports #

Supports: prompt ingress via pre_hook

/ cybersec_pre_hook

· tool-call authorization via pre_hook

, tool_hook

, and @firewall.secure_tool

· sanitized arguments and optional schema validation · delegation authorization and specialist-side one-time verification through SDK-managed state · output DLP via post_hook

/ cybersec_post_hook

with structured-output preservation · retrieved-context sanitization for Agno knowledge/RAG pipelines · streaming sanitization · production mode with stable session IDs · size limits, rate limits, retries/backoff, circuit breaker, audit logs, OpenTelemetry events, identity-rich policy context, emergency deny lists.

export AGENTICDOME_PLATFORM="agno"
export AGENTICDOME_PRODUCTION_MODE="true"
export AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD="true"
export AGENTICDOME_AGNO_MAX_INPUT_CHARS="50000"
export AGENTICDOME_AGNO_MAX_OUTPUT_CHARS="100000"
export AGENTICDOME_AGNO_MAX_TOOL_ARG_CHARS="20000"
export AGENTICDOME_AGNO_RATE_LIMIT_PER_MINUTE="120"
export AGENTICDOME_AGNO_RETRY_ATTEMPTS="2"
export AGENTICDOME_AGNO_CIRCUIT_BREAKER_FAILURES="5"
export AGENTICDOME_AGNO_AUDIT_LOGGING="true"
export AGENTICDOME_AGNO_OTEL_ENABLED="true"

Notes: environment configuration alone does not attach AgenticDome — call attach_firewall(agent_or_team)

, register create_hook_bundle()

, use the middleware/plugin helper, or assign cybersec_pre_hook

, cybersec_post_hook

, and cybersec_tool_hook

directly. Hosted/remote tools can only be protected at the local request/response boundary. Use stable session_id

/run_id

/trace_id

values. Configure the optional Redis store only when delegation authorization and execution cross workers or pods.

Official references: Agno SDK overview · Agent reference

from agenticdome_sdk.agno import (
    AgenticDomeAgnoFirewall,
    FirewallConfig,
    AgenticDomeAgnoDenied,
    DecisionTokenRecord,
    DecisionTokenStore,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
    attach_firewall,
    cybersec_pre_hook,
    cybersec_post_hook,
    cybersec_tool_hook,
    sanitize_retrieved_text,
)

Register at agent construction with ADK callback keyword arguments, attach to an existing agent, or expose as a plugin-style object for ADK plugin registration.

pip install "agenticdome-python-sdk[google-adk]"

Register callbacks — build_callback_kwargs()

returns the official callback keyword names used by LlmAgent(...)

:

from google.adk.agents import LlmAgent
from agenticdome_sdk.google_adk import AgenticDomeGoogleADKFirewall

firewall = AgenticDomeGoogleADKFirewall()

agent = LlmAgent(
    name="support_adk_agent",
    model="gemini-2.5-flash",
    instruction="Help support analysts safely.",
    **firewall.build_callback_kwargs(),
)
firewall.install_on_agent(agent)      # attach to an existing agent
plugin = firewall.create_plugin()     # plugin-style registration

Tool protection — sanitized arguments replace originals and SDK-managed security metadata is never passed to the business handler; pass a Pydantic model, Pydantic v1 model, or JSON-schema-like dict to validate arguments:

@firewall.secure_tool(tool_name="crm.customer.read", tool_platform="crm")
def lookup_customer(tool_context, args):
    return crm.get_customer(args["customer_id"])

secured_lookup = firewall.wrap_tool_handler(
    tool_name="crm.customer.read",
    tool_platform="crm",
    tool_schema={"required": ["customer_id"], "properties": {"customer_id": {"type": "string"}}},
    handler=lookup_customer,
)

Multi-agent delegation — use the public handoff methods so the adapter manages authorization state and verifies delegated execution before the specialist runs the tool:

record = await firewall.authorize_manager_handoff(
    source_agent_id="manager",
    target_agent_id="filesystem_specialist",
    target_tool_name="filesystem.read",
    target_tool_args={"path": "/reports/q4.txt"},
    tool_context=tool_context,
)

await firewall.verify_delegated_execution(
    target_agent_id="filesystem_specialist",
    tool_name="filesystem.read",
    tool_args={"path": "/reports/q4.txt"},
    tool_context=tool_context,
    decision_token=record.decision_token,
)

Google ADK capabilities, configuration, notes, and imports #

Supports: prompt screening via before_model

· model output sanitization via after_model

· tool argument authorization, schema validation, and sanitized-argument enforcement via before_tool

· tool result sanitization with structured JSON preservation via after_tool

· lifecycle audit visibility via before_agent

/ after_agent

· explicit wrappers via wrap_tool_handler()

/ @firewall.secure_tool(...)

· manager/specialist handoff authorization with SDK-managed one-time state · streaming sanitization with a sliding review buffer · rate limits, size limits, retries/backoff, circuit breaker, structured audit logs, OpenTelemetry span events, identity-rich policy context, emergency deny lists.

export AGENTICDOME_PLATFORM="google_adk"
export AGENTICDOME_GOOGLE_ADK_AGENT_ID="support_adk_agent"
export AGENTICDOME_SANITIZE_MODEL_OUTPUT="true"
export AGENTICDOME_SANITIZE_TOOL_OUTPUT="true"
export AGENTICDOME_PRODUCTION_MODE="true"
export AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD="true"
export AGENTICDOME_HANDOFF_TOKEN_TTL_S="900"
export AGENTICDOME_GOOGLE_ADK_MAX_INPUT_CHARS="50000"
export AGENTICDOME_GOOGLE_ADK_MAX_OUTPUT_CHARS="100000"
export AGENTICDOME_GOOGLE_ADK_MAX_TOOL_ARG_CHARS="20000"
export AGENTICDOME_GOOGLE_ADK_RATE_LIMIT_PER_MINUTE="120"
export AGENTICDOME_GOOGLE_ADK_RETRY_ATTEMPTS="2"
export AGENTICDOME_GOOGLE_ADK_CIRCUIT_BREAKER_FAILURES="5"
export AGENTICDOME_GOOGLE_ADK_AUDIT_LOGGING="true"
export AGENTICDOME_GOOGLE_ADK_OTEL_ENABLED="true"

Notes: register callbacks, the plugin object, or tool wrappers — env config alone does not intercept ADK execution. Use async callback methods (before_model

, after_model

, before_tool

, after_tool

) when your ADK runner supports them; the *_callback

sync methods are for synchronous configurations only. In production, provide stable ADK context values (session_id

, run_id

, trace_id

, conversation_id

, request_id

) — otherwise the adapter fails closed when AGENTICDOME_REQUIRE_STABLE_SESSION_ID_IN_PROD=true

. The SDK protects the local ADK callback boundary and returned content, not execution inside remote tools/services. Include Google Cloud identity and project context when available. Use the documented shared store and integrity secret for multi-worker or Kubernetes deployments.

from agenticdome_sdk.google_adk import (
    AgenticDomeGoogleADKFirewall,
    DecisionTokenRecord,
    DecisionTokenStore,
    FirewallConfig,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
)

Protects the local boundaries your application controls: FunctionTool functions, query calls, query-engine tools, retrieved context, and final synthesized output.

pip install "agenticdome-python-sdk[llamaindex]"

Secure a FunctionTool before giving it to a LlamaIndex agent:

from agenticdome_sdk.llamaindex import AgenticDomeLlamaIndexFirewall

firewall = AgenticDomeLlamaIndexFirewall()

def lookup_customer(customer_id: str) -> dict:
    return crm.get_customer(customer_id)

secure_lookup = firewall.to_function_tool(
    lookup_customer,
    tool_name="crm.customer.read",
    tool_platform="crm",
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

secure_lookup_fn = firewall.wrap_tool_function(
    lookup_customer,
    tool_name="crm.customer.read",
    tool_platform="crm",
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

Query and retrieval protection — around query engines you invoke directly, and at central assembly points:

answer = await firewall.run_query_securely(
    query_callable=query_engine.query,
    query_text="Find customer renewal risk.",
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

secure_query_engine = firewall.wrap_query_engine(
    query_engine,
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

secure_retriever = firewall.wrap_retriever(
    retriever,
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

safe_nodes = await firewall.sanitize_retrieval_result(
    retrieval_result=nodes,
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

node_postprocessor = firewall.create_node_postprocessor(
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

Callback visibility — global audit visibility, incident telemetry, optional extra blocking (keep hard enforcement in the wrappers; set enforce_input=True

only for an additional synchronous input check on callback query/prompt events):

from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager

handler = firewall.create_callback_handler(
    agent_id="support_llamaindex_agent",
    session_id="sess_prod_01J4X",
)

Settings.callback_manager = CallbackManager([handler])

Multi-agent handoffs — only when your application delegates from managers to specialists that can execute sensitive tools:

await firewall.authorize_manager_handoff(
    manager_agent_id="triage_manager",
    specialist_agent_id="billing_specialist",
    tool_name="billing.refund.create",
    tool_args={"invoice_id": "inv_123", "amount": 2500},
    tool_platform="billing",
    session_id="sess_prod_01J4X",
)

await firewall.verify_delegated_execution(
    specialist_agent_id="billing_specialist",
    tool_name="billing.refund.create",
    tool_args={"invoice_id": "inv_123", "amount": 2500},
    session_id="sess_prod_01J4X",
)

LlamaIndex capabilities, configuration, notes, and imports #

Supports: prompt/query screening before query execution · FunctionTool and local tool authorization · tool output review before results return to the agent · query output DLP and redaction · retrieval-result sanitization before retrieved context enters a prompt or reaches a user · query-engine and retriever wrappers for central assembly points · node postprocessor creation for RAG context sanitization · callback handler creation for global audit visibility and optional extra input blocking · optional handoff authorization and token verification · optional Redis-backed token storage for multi-worker deployments · optional creation of LlamaIndex FunctionTool

objects when LlamaIndex is installed.

export AGENTICDOME_PLATFORM="llamaindex"
export AGENTICDOME_LLAMAINDEX_AGENT_ID="support_llamaindex_agent"
export AGENTICDOME_SANITIZE_QUERY_OUTPUT="true"
export AGENTICDOME_SANITIZE_TOOL_OUTPUT="true"
export AGENTICDOME_HANDOFF_TOKEN_TTL_S="900"

Notes: wrap tools, query calls, query engines, retrievers, node postprocessors, callbacks, or output boundaries — env config alone does not intercept. LlamaIndex has many integrations and provider-native tool specs; remote services outside your process must be protected at their request/response boundary. Place wrappers in the module where components are assembled, not only inside request handlers. Use stable session_id

values, and set AGENTICDOME_REQUIRE_SESSION_ID=true

when every query/tool call must be traceable.

from agenticdome_sdk.llamaindex import (
    AgenticDomeLlamaIndexFirewall,
    DecisionTokenRecord,
    DecisionTokenStore,
    FirewallConfig,
    InMemoryDecisionTokenStore,
    RedisDecisionTokenStore,
)

For services that call Bedrock Runtime directly, stream model responses, invoke Bedrock Agents, implement action-group Lambda handlers, execute local tool-use results, or process knowledge-base retrieval. The adapter accepts any boto3-compatible client and does not import boto3 at module import time, so tests and custom clients use the same wrapper.

pip install "agenticdome-python-sdk[bedrock]"

Secure Converse — the flow is messages/system → prompt screen → Bedrock Converse → output review → sanitized response

:

import boto3
from agenticdome_sdk.aws_bedrock import AgenticDomeAWSBedrockFirewall

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
firewall = AgenticDomeAWSBedrockFirewall()

response = await firewall.converse_securely(
    bedrock_runtime_client=bedrock,
    model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=[
        {
            "role": "user",
            "content": [{"text": "Summarize this customer case."}],
        }
    ],
    agent_id="support_bedrock_agent",
    session_id="sess_prod_01J4X",
)
async for event in firewall.converse_stream_securely(
    bedrock_runtime_client=bedrock,
    model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=messages,
    agent_id="support_bedrock_agent",
    session_id="sess_prod_01J4X",
):
    yield event

Secure InvokeModel — for provider-specific payloads (Titan, Claude, Llama, Mistral, or other model-native bodies):

import json

response = await firewall.invoke_model_securely(
    bedrock_runtime_client=bedrock,
    model_id="amazon.titan-text-express-v1",
    body=json.dumps({"inputText": "Draft a customer email."}),
    agent_id="support_bedrock_agent",
    session_id="sess_prod_01J4X",
    contentType="application/json",
    accept="application/json",
)
async for event in firewall.invoke_model_with_response_stream_securely(
    bedrock_runtime_client=bedrock,
    model_id="amazon.titan-text-express-v1",
    body=json.dumps({"inputText": "Draft a customer email."}),
    agent_id="support_bedrock_agent",
    session_id="sess_prod_01J4X",
):
    yield event

The adapter extracts prompt text from common payload shapes (inputText

, prompt

, messages

, contents

, system

, Claude anthropic_version

messages, Llama/Mistral prompts, provider-native JSON bodies) and writes sanitized text back into common response fields (outputText

, completion

, generation

, answer

, text

, generated_text

, Converse output.message.content[].text

).

Bedrock Agents and action groups:

source & further reading

github.com — original article