Overview

SHAI is the secure harness around agent execution.

SHAI stands for Secure Harness AI. It is a production-grade 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 is not an agent framework. It does not replace LangChain, LangGraph, CrewAI, PydanticAI, the Anthropic SDK, or OpenAI Agents. It wraps around them. Your agent logic stays unchanged. The harness handles the security layer.

i

Most agent frameworks make tool use easy. SHAI makes tool use governable. Most observability products watch agents after the fact. SHAI sits in the execution path and acts before the boundary is crossed.

This documentation covers the full SHAI architecture, all boundary methods, the four-layer tool call gate, OWASP Agentic AI threat coverage, policy engine, sources and connectors, audit model, framework integrations, and production runtime behaviour.

The documentation is versioned alongside the codebase. If you are reading this online and want to verify you have the latest version, check the CHANGELOG.md in the repo root. Every breaking change is documented there with a migration path.

Ask your coding agent

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

example prompts
"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"
!

The skills are grounded in the actual SHAI source code and stay current as the library evolves. If you get an answer that conflicts with what you see in the code, the code is the source of truth — and please open an issue so we can update the skill.

Quick Start

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

shell
pip install shai

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.

config/harness.yaml
version: 1
tenant_id: "platform-prod"

scan_input:
  enabled: true
  block_at: high
  scanners:
    - name: regex_pii
    - name: injection_scan

scan_output:
  enabled: true
  block_at: high
  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.

config/agents/my_agent.yaml
agent_id: my_agent
allowed_tool_names:
  - search_docs
  - fetch_doc
  - send_email
allowed_capability_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.

Quick Start · LangGraph

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.

python — langgraph_agent.py
import asyncio
from harness import SHAI, Tool
from harness.core.types import Transport
from shai.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())
i

HarnessToolNode wraps all four boundary calls internally. Your LLM node never touches the harness directly — the graph handles it transparently.

Quick Start · LangChain

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.

python — langchain_agent.py
import asyncio
from harness import SHAI, Tool
from harness.core.types import Transport
from shai.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())
Quick Start · CrewAI

Same wrap_tools() pattern.

CrewAI uses the same integration module as LangChain. Wrap your tools before passing them to your agents and crew.

python — crewai_agent.py
import asyncio
from harness import SHAI, Tool
from harness.core.types import Transport
from shai.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.

Quick Start · Anthropic SDK

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.

python — anthropic_agent.py
import asyncio
import anthropic
from harness import SHAI, Tool
from harness.core.types import Transport
from shai.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"))
Quick Start · PydanticAI

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.

python — pydanticai_agent.py
import asyncio
from harness import SHAI, Tool
from harness.core.types import Transport
from shai.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())
Quick Start · OpenAI Agents

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.

python — openai_agents.py
import asyncio
from harness import SHAI, Tool
from harness.core.types import Transport
from shai.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())
Quick Start · Custom loop

Call the four boundary methods directly.

Not using a framework? SHAI is a Python library. Call the four boundary methods directly in whatever execution loop you are running.

python — custom_loop.py
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.decision == "allow":
        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"))
i

Each boundary method raises a ShaiVerdictException on block or deny. Catching this exception lets you return a safe error to the user rather than crashing the loop.

Boundaries

SHAI protects the full agent execution boundary.

BoundaryPurposeTypical 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
audit Emit one structured event per boundary call Repudiation, untraceability, unclear decision history

Prompt injection does not get treated as a text-filtering problem in isolation. It gets treated as a boundary problem across the full execution path.

Tool Gate

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.

LayerCheckBypassable?
Pre-gate Is the agent registered with the harness? No
L1 — Hard boundary Is tool_name in allowed_tool_names? No
L2 — Capability scope Do tool tags fit inside the agent or subagent capability scope? No
L3 — Policy intersection Does policy allow the call through the intersection model? By design
L4 — Sensitive args Should sensitive arguments be scanned or redacted before dispatch? Configurable
Rate limiter Is the call within the configured sliding-window budget? Configurable
!

L1 is the hard boundary. If the agent was never declared to use a tool, no policy rule can grant it later. The model can ask. SHAI can still say no. This is the property that makes the harness resilient even after the LLM context has been partially compromised.

OWASP Coverage

Controls mapped directly to OWASP Agentic AI threats.

SHAI was designed from the ground up against the OWASP Top 10 for Agentic Applications. Every threat on that list has a corresponding control in the harness. Current coverage: 8 full, 3 partial, 0 uncovered.

Threat (OWASP ID)CoveragePrimary SHAI control
Goal & instruction hijacking (T1) Partial scan_input with InjectionScanner (17-rule catalog) + L1 hard allowlist
Tool misuse (T2) Full check_tool_call L1–L4 + sliding-window rate limiter
Uncontrolled agent actions (T3) Full Subagent capability scoping + source suppression at load_agent()
Resource overload (T4) Full Sliding-window rate limiter — global and per-tool budgets
Direct prompt injection (T5) Full Input scanning with YAML rule catalogs — block, alert, or redact per severity
Indirect prompt injection (T6) Full Tool-result scanning with 9-rule document-focused pattern catalog
Repudiation & untraceability (T8) Full One HMAC-SHA256-signed audit event per boundary call, SIEM-safe JSONL
Privilege escalation (T9) Full L2 capability gate + policy intersection model enforcing least privilege
Sensitive data exposure (T11) Full RegexPII scanning on input, tool args, and output — Luhn-validated card matching
Data exfiltration (T16) Partial Application-layer output PII scanning. Network-layer via shai-connectivity (roadmap)
Supply chain compromise (T17) Partial FileScanner + MIME validation + macro detection + source suppression + secret:// URIs

Goal & instruction hijacking (T1)

The InjectionScanner runs a 17-rule YAML catalog covering instruction override, role spoofing, delimiter smuggling, and policy evasion — before the LLM processes a single token. Even if injection reaches the model, the L1 gate prevents any undeclared tool from executing.

Tool misuse (T2)

L1 enforces the declared allowlist before policy runs. L2 gates capability tags. L3 evaluates the intersection of subagent, parent, and global policy rules — allow, deny, redact, or suppress. L4 scans sensitive arguments before dispatch. A sliding-window rate limiter runs in parallel to contain flooding.

Uncontrolled agent actions (T3)

Agents declare which tools they can use and which tags are in scope. Subagents inherit a scoped subset of their parent — a read-only research agent can never acquire write access, regardless of what the LLM requests. Source suppression prevents undeclared MCP servers from activating at runtime.

Resource overload (T4)

Two independent counters — a global call budget and a per-tool budget — must both pass before dispatch. The window size and max calls are fully configurable. Abusive loops and runaway agents hit a hard ceiling without disrupting normal turns.

Direct prompt injection (T5)

The InjectionScanner evaluates 17 YAML rules covering jailbreak patterns, context switching, encoded payloads, homoglyph obfuscation, tool coercion, and more. Each rule has a severity. You choose the threshold to block, alert, or redact — per boundary, per scanner.

Indirect prompt injection (T6)

This is the checkpoint most harnesses miss entirely. Documents, search results, and API payloads are scanned with a document-tuned 9-rule pattern catalog before the LLM is allowed to ingest them — closing the exact attack vector used in ClawJacked-style exploits.

Repudiation & untraceability (T8)

Events are HMAC-SHA256 signed and tamper-evident. No raw user text, matched substrings, or LLM output ever appears in the log — making it safe to forward directly to a SIEM. When something goes wrong, you know exactly what the agent did, step by step.

Privilege escalation (T9)

The L2 capability gate and policy intersection model keep parent rules, subagent rules, and global rules aligned around least privilege at runtime. A subagent spawned for read-only research cannot acquire write access — not through the LLM, not through policy, not at all.

Sensitive data exposure (T11)

RegexPIIScanner detects email, phone, SSN, and credit card patterns using Luhn-validated matching. Each scanner is configurable to block, alert, or redact in-place. Matched values never appear in the audit log — only the category name. Your agent should never accidentally surface a credential it encountered along the way.

Data exfiltration (T16)

Application-layer PII detection catches accidental credential surfacing today. Deeper network-layer enforcement — controlling which outbound connections agents can make — is on the roadmap as shai-connectivity, extending coverage to covert exfiltration channels.

Supply chain compromise (T17)

FileScanner performs structural checks with MIME validation, macro detection, and nested archive inspection. Source suppression prevents untrusted sources from loading. Credentials are resolved via secret:// URIs — never hardcoded, never logged. Connector manifests for Slack, GitHub, Gmail, Jira, Notion, Stripe, PostgreSQL, and Google Drive ship with sane security defaults out of the box.

Application-layer coverage is active today. Deeper network-layer enforcement for exfiltration and broader supply-chain control is planned through the shai-connectivity extension.

Policy

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 all hard gates (L1 and L2) have already passed.

Rule schema

policy rules
- 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, or source_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.
Sources

Local tools, skill tools, MCP servers, and connector manifests under one model.

Sources are activated at load_agent() time, not per turn. Required MCP sources fail safe at load — they raise a ConfigError rather than silently proceeding with a broken configuration. The harness gates the call; dispatch remains in the agent layer after approval.

Source typeUse
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
Connector manifests Slack, GitHub, Notion, Jira, Gmail, PostgreSQL, Stripe, Google Drive — pre-configured with sane security defaults
sources in agent config
sources:
  - name: slack
    connector: slack
  - name: github
    connector: github
  - name: docs_local
    transport: local
    tool_names: [search_docs, fetch_doc]
  - name: internal_mcp
    transport: mcp
    url: "http://localhost:8080/mcp"
    required: true  # raises ConfigError if unreachable at load time

Credentials for connector sources are resolved via secret:// URIs — never hardcoded in config files, never written to the audit log. Only the category name appears in log output.

Audit

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=deny always carries a deny_reason.
  • Optional HMAC-SHA256 signing makes tampering evident.
  • Sinks fan out concurrently and produce JSONL-compatible output for SIEM pipelines.
  • tenant_id comes from harness configuration, not the caller — it cannot be spoofed per turn.
  • Available sinks: stdout, file (JSONL), custom (implement the sink interface).
example audit event — deny
{
  "boundary":    "tool_call_gate",
  "decision":    "deny",
  "agent_id":    "orchestrator_agent",
  "sub_agent_id":"research_sub",
  "tool_name":   "send_email",
  "deny_reason": "research_sub is read-only",
  "tenant_id":   "platform-prod",
  "ts":          "2026-06-26T09:14:02.311Z",
  "hmac":        "sha256=a3f9..."
}
shell — tail audit log
shai audit tail --file logs/audit.jsonl --follow
shai audit tail --decision deny --boundary tool_call_gate
Integrations

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.

FrameworkModuleIntegration pattern
LangGraph shai.integrations.langgraph Swap ToolNodeHarnessToolNode
LangChain shai.integrations.langchain wrap_tools(tools, harness, ctx)
CrewAI shai.integrations.langchain wrap_tools(tools, harness, ctx)
Anthropic SDK shai.integrations.anthropic_sdk gated_dispatch(harness, ctx, block, registry)
PydanticAI shai.integrations.pydanticai @harness_tool decorator or add_harness_middleware()
OpenAI Agents shai.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.

Production

Built for production agent loops.

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.
  • Stateless per turn — all agent state lives in the ctx object loaded once at startup.
  • 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 ConfigError at startup, never silently at runtime.
  • Python 3.9+ — no pinned upper bound on the Python version.
i

For high-throughput deployments, benchmark scanner concurrency with your expected tool result payload sizes. The harness is designed for agent turn latencies — not sub-millisecond streaming — but the concurrent scanner model keeps overhead proportional to payload complexity, not turn volume.

© SHAI by AIBestLabs.com. All rights reserved.