SHAI Documentation
SHAI is the secure harness around agent execution.
SHAI stands for Secure Harness AI. It is a Python security library for AI agents. It sits between your agent code and tool dispatch, scanning every boundary that matters and enforcing policy before actions become incidents.
SHAI sits in the execution path and decides at each boundary, so the operator's config governs what an agent can do on every call.
This documentation covers the full SHAI architecture, all boundary methods, the seven-layer tool call gate, OWASP Agentic AI threat coverage, policy engine, tool sources and MCP manifests, audit model, framework integrations, and production runtime behaviour.
You don't need to read the full documentation to start building.
The SHAI repository ships with a set of Claude Skills — structured knowledge files that give your AI coding agent (Claude Code, Cursor, Windsurf, Copilot, or any MCP-compatible tool) deep, up-to-date knowledge of SHAI's architecture, APIs, and integration patterns.
Once you have the repo, just ask your coding agent naturally. It already knows SHAI.
The skills live at .claude/skills/ in the repo root. Your AI coding agent picks
them up automatically if it supports Claude Skills or project-level context. No setup required.
Things you can ask your coding agent
"Wrap my existing LangGraph agent with SHAI" "Write a harness.yaml config that blocks all external writes and allows read-only tools" "Add indirect prompt injection protection to my tool result loop" "How do I configure a subagent that can only use read-tagged tools?" "Generate a policy rule that denies all MCP calls unless explicitly approved" "Set up HMAC-signed audit logging with a file sink" "What's the difference between L1, L2, and L3 in the tool gate?" "Show me how to use gated_dispatch with the Anthropic SDK" "Write a SHAI integration for my CrewAI agent that uses the GitHub connector"
Install, configure, and protect your agent in four steps.
SHAI is designed for one instance per deployment, shared safely across many concurrent turns. Start here, then jump to the template for your framework below.
Step 1 — Install
pip install shai-harness # optional — clone to run the tests and examples git clone https://github.com/fad-schme/SHAI.git cd SHAI pytest tests/unit -q
Step 2 — Create config/harness.yaml
This file is your security policy. Each agent gets its own config file. The harness loads it once at startup.
version: 1
tenant_id: "platform-prod"
scan_input:
enabled: true
block_at: high
on_error: fail_closed
scanners:
- name: injection_scan
- name: jailbreak_scan
- name: identity_spoof_scan
- name: regex_pii
scan_tool_result:
enabled: true
block_at: high
on_error: fail_closed
scanners:
- name: injection_scan
- name: identity_spoof_scan
- name: jailbreak_scan
scan_mcp_metadata:
enabled: true
block_at: medium
scanners:
- name: mcp_metadata_scan
scan_output:
enabled: true
block_at: high
on_error: fail_closed
scanners:
- name: regex_pii
policy:
rules:
- id: allow_read_tools
match:
tool_tags: [read]
action: allow
- id: deny_external_write
match:
tool_tags: [external_write]
action: deny
reason: "external_write requires explicit approval"
audit_sinks:
- name: stdout
Step 3 — Create config/agents/my_agent.yaml
Each agent declares exactly which tools it is allowed to use. This becomes the L1 hard boundary — no policy rule can grant access to a tool not in this list.
id: my_agent
allowed_tool_names:
- search_docs
- fetch_doc
- send_email
allowed_tags:
- read
- internal
sources:
- name: docs_local
transport: local
tool_names: [search_docs, fetch_doc]
Step 4 — Pick your framework template
Choose the template below that matches your stack. Each one shows a complete, working integration from harness instantiation to boundary calls.
Replace ToolNode with HarnessToolNode.
One-line swap. All existing graph nodes, edges, and conditional logic stay exactly as they are. SHAI gates every tool transition in the graph automatically.
import asyncio from harness import SHAI, Tool from harness.core.types import Transport from harness.integrations.langgraph import HarnessToolNode from langgraph.graph import StateGraph, END # 1. Load harness and tools async def build_graph(): harness = await SHAI.from_yaml("config/harness.yaml") tools = [ Tool(name="search_docs", tags=["read", "internal"], transport=Transport.LOCAL), Tool(name="fetch_doc", tags=["read", "internal"], transport=Transport.LOCAL), ] await harness.register_tools(tools) ctx = await harness.load_agent("config/agents/my_agent.yaml") # 2. Build graph — swap ToolNode for HarnessToolNode builder = StateGraph(dict) builder.add_node("agent", your_llm_node) builder.add_node("tools", HarnessToolNode(tools, harness, ctx)) # ← only change builder.add_edge("tools", "agent") builder.add_conditional_edges("agent", should_call_tools, {"tools": "tools", END: END}) builder.set_entry_point("agent") return builder.compile() graph = asyncio.run(build_graph())
HarnessToolNode wraps all four boundary calls internally. Your LLM node never touches the harness directly — the graph handles it transparently.
Wrap your tools with wrap_tools().
Pass your existing tool list through wrap_tools(). Use the returned list exactly as you would the originals — in chains, agents, or tool-calling runnables.
import asyncio from harness import SHAI, Tool from harness.core.types import Transport from harness.integrations.langchain import wrap_tools from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_anthropic import ChatAnthropic async def main(): harness = await SHAI.from_yaml("config/harness.yaml") raw_tools = [search_docs_tool, fetch_doc_tool] # your existing LangChain tools await harness.register_tools([ Tool(name="search_docs", tags=["read"], transport=Transport.LOCAL), Tool(name="fetch_doc", tags=["read"], transport=Transport.LOCAL), ]) ctx = await harness.load_agent("config/agents/my_agent.yaml") # Wrap — use secure_tools everywhere you used raw_tools secure_tools = wrap_tools(raw_tools, harness, ctx) llm = ChatAnthropic(model="claude-sonnet-4-6") agent = create_tool_calling_agent(llm, secure_tools, prompt) executor = AgentExecutor(agent=agent, tools=secure_tools) result = await executor.ainvoke({"input": "Find the onboarding guide"}) print(result) asyncio.run(main())
Same wrap_tools() pattern.
CrewAI uses the same integration module as LangChain. Wrap your tools before passing them to your agents and crew.
import asyncio from harness import SHAI, Tool from harness.core.types import Transport from harness.integrations.langchain import wrap_tools # same module for CrewAI from crewai import Agent, Task, Crew async def main(): harness = await SHAI.from_yaml("config/harness.yaml") await harness.register_tools([ Tool(name="search_docs", tags=["read"], transport=Transport.LOCAL), Tool(name="send_report", tags=["external_write"], transport=Transport.LOCAL), ]) ctx = await harness.load_agent("config/agents/my_agent.yaml") secure_tools = wrap_tools([search_docs_tool, send_report_tool], harness, ctx) researcher = Agent( role="Research Analyst", goal="Find and summarise relevant documents", tools=secure_tools, llm="claude-sonnet-4-6" ) task = Task(description="Research the Q3 performance report", agent=researcher) crew = Crew(agents=[researcher], tasks=[task]) result = crew.kickoff() print(result) asyncio.run(main())
Each agent in a multi-agent crew should use a separate ctx loaded from its own agent YAML. This ensures subagent capability scoping is enforced independently per crew member.
Slot gated_dispatch() into your tool loop.
If you are running a raw Anthropic SDK tool loop, replace your manual tool dispatch with gated_dispatch(). Everything else in the loop stays the same.
import asyncio import anthropic from harness import SHAI, Tool from harness.core.types import Transport from harness.integrations.anthropic_sdk import gated_dispatch client = anthropic.Anthropic() async def run_agent(user_message: str): harness = await SHAI.from_yaml("config/harness.yaml") await harness.register_tools([ Tool(name="search_docs", tags=["read"], transport=Transport.LOCAL), ]) ctx = await harness.load_agent("config/agents/my_agent.yaml") # Scan input before the LLM sees it await harness.scan_input(user_message, ctx) messages = [{"role": "user", "content": user_message}] while True: response = client.messages.create( model="claude-sonnet-4-6", max_tokens=4096, tools=tool_definitions, messages=messages, ) if response.stop_reason == "end_turn": final = response.content[0].text await harness.scan_output(final, ctx) # scan before returning return final # For each tool use block, replace manual dispatch with gated_dispatch tool_results = [] for block in response.content: if block.type == "tool_use": result = await gated_dispatch(harness, ctx, block, tool_registry) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": result, }) messages += [{"role": "assistant", "content": response.content}, {"role": "user", "content": tool_results}] asyncio.run(run_agent("Find the onboarding guide"))
Decorator for individual tools, middleware for full pipeline coverage.
Both patterns are supported. Use @harness_tool to gate specific tools selectively, or add_harness_middleware() to cover all boundaries automatically.
import asyncio from harness import SHAI, Tool from harness.core.types import Transport from harness.integrations.pydanticai import harness_tool, add_harness_middleware from pydantic_ai import Agent async def main(): harness = await SHAI.from_yaml("config/harness.yaml") await harness.register_tools([ Tool(name="search_web", tags=["read", "external"], transport=Transport.LOCAL), ]) ctx = await harness.load_agent("config/agents/my_agent.yaml") # Option A: decorate individual tools @harness_tool(harness, ctx) async def search_web(query: str) -> str: return perform_search(query) # Option B: add middleware to the agent for full pipeline coverage agent = Agent("claude-sonnet-4-6", tools=[search_web]) add_harness_middleware(agent, harness, ctx) result = await agent.run("What is the latest news on AI security?") print(result.data) asyncio.run(main())
Hook and wrap — two patterns, both supported.
Use make_before_tool_hook() to intercept every tool call in an agent run, or wrap_tool() to gate individual tools selectively.
import asyncio from harness import SHAI, Tool from harness.core.types import Transport from harness.integrations.openai_agents import make_before_tool_hook, wrap_tool from agents import Agent, Runner, RunHooks, function_tool async def main(): harness = await SHAI.from_yaml("config/harness.yaml") await harness.register_tools([ Tool(name="search_docs", tags=["read"], transport=Transport.LOCAL), ]) ctx = await harness.load_agent("config/agents/my_agent.yaml") # Option A: before_tool_call hook gates every tool in this run hooks = RunHooks(before_tool_call=make_before_tool_hook(harness, ctx)) # Option B: wrap individual tools @function_tool async def search_docs(query: str) -> str: return perform_search(query) secure_search = wrap_tool(search_docs, harness, ctx) agent = Agent(name="ResearchAgent", tools=[secure_search]) result = await Runner.run(agent, "Find the architecture overview", hooks=hooks) print(result.final_output) asyncio.run(main())
Call the five boundary methods directly.
Not using a framework? SHAI is a Python library. Call the five boundary methods directly in whatever execution loop you are running. (scan_file is optional; the four shown below cover most tool-loop shapes.)
import asyncio from harness import SHAI, Tool from harness.core.types import Transport async def agent_loop(user_message: str): harness = await SHAI.from_yaml("config/harness.yaml") await harness.register_tools([ Tool(name="search_docs", tags=["read", "internal"], transport=Transport.LOCAL), Tool(name="send_email", tags=["external_write"], transport=Transport.LOCAL), ]) ctx = await harness.load_agent("config/agents/my_agent.yaml") # Boundary 1 — scan input before the LLM sees it await harness.scan_input(user_message, ctx) # ... call your LLM here ... tool_name = "search_docs" tool_args = {"query": "onboarding guide"} # Boundary 2 — gate every tool call before dispatch gate = await harness.check_tool_call(tool_name, tool_args, ctx) if gate.allowed: tool_result = your_tool_registry[tool_name](**tool_args) # Boundary 3 — scan tool return before re-entering LLM context await harness.scan_tool_result(tool_result, ctx) # ... call LLM again with tool result ... final_response = "The onboarding guide is at /docs/onboarding.md" # Boundary 4 — scan output before it reaches the user await harness.scan_output(final_response, ctx) return final_response asyncio.run(agent_loop("Find the onboarding guide"))
Boundary methods return verdicts. They do not raise on a policy decision. Check verdict.blocked (scans) and gate.allowed (tool gate) explicitly and route to a safe response. Exceptions from a boundary method mean something structural is wrong — unknown agent, unknown tool, config error — and should be caught as HarnessError.
SHAI protects the full agent execution boundary.
| Boundary | Purpose | Typical threats addressed |
|---|---|---|
scan_input |
Inspect user text before it reaches the LLM | Goal hijacking, direct prompt injection, PII exposure |
check_tool_call |
Gate tool execution before dispatch | Tool misuse, uncontrolled actions, privilege escalation, resource overload |
scan_tool_result |
Inspect returned content before it re-enters model context | Indirect prompt injection, hostile documents, poisoned search results |
scan_output |
Inspect model output before it reaches the user | Sensitive data exposure, application-layer exfiltration |
scan_file |
Inspect uploaded files at ingress (opt-in) | Malicious PDFs, Office macros, archive bombs, SVG scripts, EXIF anomalies |
scan_mcp_metadata |
Inspect tool names, descriptions, and argument schemas at MCP connection time | Tool-description injection, metadata shadowing |
Each boundary sits on the execution path, so a payload is evaluated at every surface it crosses.
Scanner pipeline
Every scan boundary runs three layers automatically:
- Heuristic scanner — always on, not configurable. Detects structural anomalies: high-entropy segments (obfuscated payloads), instruction-dense text, register shifts (business English → command syntax), and embedded LLM markup (
<|system|>,[INST]). - Configured scanners — declared per boundary in
harness.yaml. Available:injection_scan,jailbreak_scan,identity_spoof_scan,regex_pii,command_injection_scan,mcp_metadata_scan. The injection, jailbreak, and identity-spoof catalogs each ship with French, Spanish, German, and Mandarin variants. - Ensemble promotion — severity is promoted when scanners from distinct method families agree on the same finding category.
Cross-boundary signals
scan_input attaches a TurnSignals bus to the context; later boundaries read and write it. check_tool_call uses it at L6, scan_tool_result tightens block_at by one level when input flagged injection, and scan_output computes a consolidated turn risk that blocks at 0.60 before clearing the bus. A separate cross-turn accumulator scores escalation spread across a session.
Error handling
The on_error field on each scan boundary controls what happens when a scanner fails:
fail_closed(default) — scanner failure → BLOCK.fail_open— scanner failure → empty findings, pipeline continues.degrade— scanner failure → WARN, audit event flagged.
A per-scanner circuit breaker prevents repeated calls to broken adapters. Failures emit boundary=system audit events.
Every tool call has to earn the right to execute.
check_tool_call is mandatory and cannot be disabled. Strict order matters: each layer runs only if the previous one passed.
| Layer | Check | Bypassable? |
|---|---|---|
| Pre-gate | Revocation, rate limit, session budget, MCP manifest baseline approval, agent registered. | No |
| L1 — Hard allowlist | Is tool_name in allowed_tool_names? |
No |
| L2 — Argument rules | Deterministic parameter constraints (max_value, min_value, allowlist, pattern, required). |
No |
| L3 — Irreversibility gate | Tools marked SENSITIVE or IRREVERSIBLE require a quorum of signed ApprovalGrants on ctx.approvals. |
No |
| L4 — Capability tags | Do tool tags fit inside the agent or subagent capability scope? | No |
| L5 — Policy intersection | Subagent → parent → global rules. First deny anywhere wins. | By design |
| L6 — Signal correlation | Reads TurnSignals from earlier boundaries. Input flagged for injection + write-capable tool → deny. |
No |
| L7 — Argument scan | Runs for sensitive-tagged tools, or when L6 tightened. |
Configurable |
L1 is the hard boundary. A tool that is not in allowed_tool_names is reachable by no policy rule, override, or request from the model.
Approvals
Each grant is bound to (agent_id, tenant_id, tool_name, args_digest, approver_id, expiry) and verified offline against an HMAC secret, so the gate checks a signature rather than calling an authorization server. Quorum counts distinct approver_ids and is configured separately for sensitive and irreversible tools. Grants do not propagate to subagents.
check_tool_call:
approvals:
secret: "secret://SHAI_APPROVAL_KEY"
sensitive_quorum: 1
irreversible_quorum: 2
Dispatch tokens
Every allowed decision issues a signed, single-use token bound to the tool, agent, and source, with a short TTL. ShaiTransport validates it on each outbound MCP request; a local tool validates it with verify_tool_dispatch inside dispatch_scope. connectivity.token_policy: strict is the default and refuses an untokened request.
Revocation
Revoking an agent denies it at the pre-gate within cache_ttl_seconds, leaves every other agent running, and persists across restarts.
shai agents revoke my_agent --config config/harness.yaml shai agents restore my_agent --config config/harness.yaml shai agents revocations --config config/harness.yaml
Controls mapped directly to OWASP Agentic AI threats.
SHAI is designed against the OWASP Top 10 for Agentic Applications (2026). Each entry gives the threat as OWASP scopes it, the coverage rating, and the control that answers it.
| Threat | Coverage | Primary SHAI control |
|---|---|---|
| ASI01 · Agent Goal Hijack | Partial | scan_input, scan_file, and scan_tool_result run the injection, jailbreak, and identity-spoof catalogs plus the heuristic scanner |
| ASI02 · Tool Misuse and Exploitation | Full | check_tool_call L1–L7, with revocation, rate limits, session budget, and manifest approval at the pre-gate |
| ASI03 · Identity & Privilege Abuse | Partial | Subagent subset enforcement at load_agent(), L4 capability tags, L5 policy intersection, secret:// credential resolution |
| ASI04 · Agentic Supply Chain Vulnerabilities | Partial | MCP manifest baseline approval at connect and on every call, scan_mcp_metadata, signed pattern bundles, startup attestation |
| ASI05 · Unexpected Code Execution | Limited | command_injection_scan at any boundary, including tool arguments at the gate |
| ASI06 · Memory & Context Poisoning | Partial | Ingress scanning on every received surface; memory writes pass check_tool_call like any other action |
| ASI07 · Insecure Inter-Agent Communication | Out of scope | Compose with a message layer that authenticates peers |
| ASI08 · Cascading Failures | Partial | on_error: fail_closed, rate limits, session budgets, loop detection, persistent revocation |
| ASI09 · Human-Agent Trust Exploitation | Out of scope | Approval quorum raises the cost of a single approver |
| ASI10 · Rogue Agents | Full | Registered-agent admission, config-gated calls, revocation, consolidated turn-risk block at scan_output |
ASI01 · Agent Goal Hijack
Every entry point is scanned before content reaches the LLM. The injection, jailbreak, and identity-spoof catalogs run at scan_input, scan_file, and scan_tool_result, each merged with its French, Spanish, German, and Mandarin variants, over de-obfuscated views of the text. scan_file adds a document-tuned catalog. A cross-turn accumulator scores escalation spread across turns, and the signed pattern DB extends the catalogs without a code release.
ASI02 · Tool Misuse and Exploitation
An agent does what the operator's config allows, enforced on every call. L1 checks the declared allowlist, L2 deterministic argument rules, L3 the approval quorum for sensitive and irreversible tools, L4 capability tags, L5 the intersection of subagent, parent, and global policy, L6 cross-boundary signals, L7 argument scanning. Revocation, rate limits, session budgets, and MCP manifest approval run ahead of all seven. First deny wins.
ASI03 · Identity & Privilege Abuse
A subagent's allowed_tool_names and allowed_tags must be subsets of its parent's, checked at load_agent() and re-checked by L4 on every call. L5 intersects parent and subagent policy rules. check_tool_call admits calls only from agents loaded through SHAI.load_agent(). Credentials resolve from secret:// URIs at load.
ASI04 · Agentic Supply Chain Vulnerabilities
An MCP source is declared by name under sources: and its manifest resolves by convention from mcp_manifests_dir. The source is built when the manifest hash matches an approved baseline, and that baseline is re-checked at connect and on every check_tool_call for the source. Tool names, descriptions, and tags come from the approved manifest and pass scan_mcp_metadata. Pattern bundles are HMAC-signed and verified row by row before they apply. from_yaml() emits a startup attestation of every adapter, manifest, and policy digest it wired.
ASI05 · Unexpected Code Execution
command_injection_scan parses shell syntax and flags compositions with no benign reading, such as a pipeline whose sink is an interpreter, a redirect to /dev/tcp, or a fetch composed with an exec. It runs at any boundary, including tool arguments at check_tool_call. Requires the shell extra.
ASI06 · Memory & Context Poisoning
Everything that reaches memory has crossed a SHAI boundary. User input, uploads, and tool and retrieval results are scanned before the LLM ingests them, with the controls in ASI01. Anything the agent then writes to memory is a tool call through check_tool_call.
ASI08 · Cascading Failures
Scanner errors follow on_error, which defaults to fail_closed. SessionBudget enforces max_steps and max_tool_calls_per_prompt, and loop detection denies a call whose fingerprint falls within loop_similarity_threshold of recent calls. RateLimiter applies global and per-tool windows. Revocation stops one agent at the pre-gate while the others keep running. Budgets and rate limits are held per SHAI instance.
ASI10 · Rogue Agents
A rogue agent stays bound by its config. Calls are admitted only from registered agents and gated against the operator's config on every turn. Revocation takes effect within cache_ttl_seconds and survives a restart. A consolidated turn-risk score at scan_output blocks turns whose combined cross-boundary signals cross 0.60.
Controls outside the OWASP list
These do not map to a single OWASP entry. Same set as the landing page.
| Control | Coverage | Mechanism |
|---|---|---|
| Dispatch-bound execution | Full | Signed single-use token per allowed call, validated by ShaiTransport for MCP and verify_tool_dispatch for local tools |
| Human approval for irreversible actions | Full | L3 quorum of signed ApprovalGrants, verified offline against an HMAC secret |
| Tamper-evident audit trail | Full | One HMAC-SHA256-signed event per boundary call, no raw content in any field |
| Sensitive data handling | Full | regex_pii at input, output, file content, and tool arguments, with block, alert, or redact per boundary |
| Data exfiltration at egress | Partial | scan_output on the full response before delivery |
Cross-cutting · data and audit
Audit events carry finding_count, max_severity, boundary, decision, adapters, and structured metadata, HMAC-SHA256 signed with an operator-supplied secret. PII and secret categories are scanned at input, output, file content, and tool arguments, with redaction applied before text leaves the boundary.
Policy runs before the action, not after the incident.
SHAI uses an intersection model: subagent rules and parent rules are evaluated first, then global rules. First match wins. No match defaults to allow, but only after the hard gates (L1 allowlist and L4 capability tags) have already passed.
Rule schema
- id: deny_mcp_by_default
match:
transport: [mcp]
action: deny
reason: "MCP requires explicit permission"
- id: allow_read_tools
match:
tool_tags: [read]
action: allow
- id: redact_sensitive_arg
match:
tool_names: [send_email]
action: redact
redact:
body: "[REDACTED]"
- id: deny_subagent_external_write
match:
subagent_id: research_sub
tool_tags: [external_write]
action: deny
reason: "research_sub is read-only"
- Actions:
allow,deny,redact,suppress. - Match by
tool_name,tool_tags,transport,agent_id,subagent_id, orsource_tags. - Policy cannot override the L1 hard capability boundary. That gate fires before policy runs.
- Subagent rules, parent rules, and global rules run through the intersection model in that order.
- Transport-aware policy lets teams treat MCP sources differently from local or skill tools.
Local tools, skill tools, and MCP servers under one model.
Sources are activated at load_agent() time, not per turn. A required source that fails to load raises a ConfigError at startup. The harness gates the call and dispatch stays in the agent layer.
| Source type | Use |
|---|---|
LocalSource |
Registered Python-callable tools in the local runtime |
SkillSource |
Curated subsets of registered tools under a distinct transport class |
MCPSource |
MCP servers loaded through SSE and JSON-RPC tool catalogs |
| MCP manifests | Slack, GitHub, Notion, Jira, Gmail, PostgreSQL, Stripe, Google Drive, shipped with url, allow-lists, tags, and per-tool specs |
sources: # MCP sources are declared by name. Everything else comes from the # manifest at/ .yaml - name: slack transport: mcp - name: github transport: mcp required: false - name: docs_local transport: local tool_names: [search_docs, fetch_doc] tags: [internal]
Onboarding an MCP manifest
A manifest's hash enters the baseline store through shai mcp onboard. The source is built when the hash matches that baseline, which is re-checked at connect and on every tool call for the source.
shai mcp onboard mcp_manifests/slack.yaml --config config/harness.yaml
Credentials resolve from secret:// URIs at from_yaml() time and are never written to config files or the audit log.
Every boundary leaves evidence.
Exactly one structured audit event is emitted for every boundary call, regardless of outcome. Allowed, denied, blocked, redacted, suppressed, disabled — it is visible.
- No raw prompts, outputs, tool arguments, or matched secrets are written into audit fields.
decision=denyalways carries adeny_reason.- Optional HMAC-SHA256 signing makes tampering evident.
- Sinks fan out concurrently and produce JSONL-compatible output for SIEM pipelines.
tenant_idcomes from harness configuration, not the caller — it cannot be spoofed per turn.- Available sinks:
stdout,file(JSONL),custom(implement the sink interface).
{
"timestamp": "2026-06-26T09:14:02.311000+00:00",
"boundary": "tool_call_gate",
"decision": "deny",
"duration_ms": 2,
"tenant_id": "platform-prod",
"agent_id": "orchestrator_agent",
"sub_agent_id": "research_sub",
"tool_name": "send_email",
"transport": "local",
"adapters": ["rules"],
"finding_count": 0,
"deny_reason": "research_sub is read-only",
"audit_tags": {"team": "platform", "env": "prod"},
"signature": "a3f9..."
}
Startup attestation
SHAI.from_yaml() emits one boundary=system, decision=startup event recording what the process wired: each scanner, sink, and policy adapter with the SHA256 of its defining source file, MCP manifest digests, the pattern-DB and policy digests, and every declared source with its URL stripped of credentials. shai harness inspect shows the same topology offline.
Network events
ShaiTransport emits a NetworkAuditEvent per outbound MCP request to the same sinks under event_type network_egress. It shares token_id with the gate event for the same decision, so the two join in a SIEM query.
shai audit tail --file logs/audit.jsonl --follow shai audit tail --decision deny --boundary tool_call_gate
Use SHAI with the frameworks your team already runs.
All integration modules are lazy-loaded. Importing the LangGraph integration does not pull in LangChain. SHAI adds no transitive dependencies beyond what your project already has.
| Framework | Module | Integration pattern |
|---|---|---|
| LangGraph | harness.integrations.langgraph |
Swap ToolNode → HarnessToolNode |
| LangChain | harness.integrations.langchain |
wrap_tools(tools, harness, ctx) |
| CrewAI | harness.integrations.langchain |
wrap_tools(tools, harness, ctx) |
| Anthropic SDK | harness.integrations.anthropic_sdk |
gated_dispatch(harness, ctx, block, registry) |
| PydanticAI | harness.integrations.pydanticai |
@harness_tool decorator or add_harness_middleware() |
| OpenAI Agents | harness.integrations.openai_agents |
make_before_tool_hook() or wrap_tool() |
You do not need to rewrite your agent to make it governable. SHAI wraps the tool boundary where the risk actually happens. See the Quick Start section above for a working code template for each framework.
Developer tools from the command line.
Installed with the package. All commands run through the shai entrypoint.
Config validation
# Validate harness.yaml and all agent files shai validate --config config/harness.yaml --agents-dir config/agents/
Agent management
# List agents with tool count, subagents, and sources shai agents list --agents-dir config/agents/ # Kill switch shai agents revoke my_agent --config config/harness.yaml shai agents restore my_agent --config config/harness.yaml shai agents revocations --config config/harness.yaml
Topology
# What a config wires up, offline shai harness inspect --config config/harness.yaml --agents-dir config/agents/ # agent -> source -> tool -> tag as a graph shai harness graph --config config/harness.yaml --agents-dir config/agents/ | dot -Tsvg -o topology.svg
MCP manifests
# Record a manifest hash in the baseline store shai mcp onboard mcp_manifests/slack.yaml --config config/harness.yaml
Audit log
# Tail the live audit log shai audit tail --file logs/audit.jsonl --follow # Filter to gate denials only shai audit tail --file logs/audit.jsonl --boundary tool_call_gate --decision deny # Show the last 50 events shai audit tail --file logs/audit.jsonl --last 50
Pattern management
# Apply a signed pattern bundle shai patterns apply --bundle rules.json --db state/patterns.db --secret SIGNING_KEY # List all rules in the database shai patterns list --db state/patterns.db # Verify all signatures shai patterns verify --db state/patterns.db --secret SIGNING_KEY
Heuristic candidates
# List candidates — most frequent first shai patterns candidates --db state/patterns.db # Filter by status shai patterns candidates --db state/patterns.db --status open # Promote a candidate (enters scan read path) shai patterns promote --db state/patterns.db --id 12 # Dismiss a false positive shai patterns dismiss --db state/patterns.db --id 8 # Retire after writing a regex rule shai patterns retire --db state/patterns.db --id 12
Candidate skeletons show attack structure without raw text: ··· [INST] ··· ignore override ··· {"role":"system"} ··· call send_email ···. This gives engineers enough context to evaluate and write targeted regex rules.
Designed for production runtime behaviour.
SHAI is designed to stay out of the way until a boundary matters. The hot path is read-heavy and lock-free where possible; all expensive setup happens at load_agent() time.
- One instance per deployment — shared safely across many concurrent turns.
- One context per conversation. Derive it with
agent.for_conversation(id), since the context carries that turn's signal bus. - No lazy initialization on the hot path — tools, sources, policy, scanners, and audit sinks are resolved once before traffic arrives.
- Concurrent scanners — multiple scanners at the same boundary run concurrently, not serially.
- Concurrent audit sinks — fan-out to multiple sinks happens concurrently.
- O(1) amortised rate limiter — sliding-window deque pruning does not grow with traffic volume.
- Fail-safe config loading — a misconfigured or unreachable required source raises a
ConfigErrorat startup, never silently at runtime. - Incremental pattern updates — supplemental injection patterns stored in a signed SQLite database, verified at startup, distributed via
shai patterns apply. No package rebuild needed. - Circuit breakers. Five consecutive scanner failures open the breaker for 60 s, with backoff on repeated tripping.
- Persistent revocation. The kill switch is file-backed and survives a restart.
- Buffer before
scan_output. The boundary scans the response as a unit, so deliver it after the verdict returns. - Python 3.11+ — no pinned upper bound on the Python version.