Firewall SDK

A runtime security firewall for LLM agents.

New to AgentHacker? Start with docs/getting_started.md: it covers the high-level Firewall client (the ~10-minute "Tier 1" path) and a feature reference with costs and recommendations. This README is the deeper "Tier 2" reference — the four local scan checkpoints you wire into an agent loop.

The SDK provides four checkpoint scan functions that agents call at specific points in their agent loop. The SDK owns scanning and detection. The agent owns everything else: the loop, the tools, the auth, the prompts, the data.

Installation

The distribution name is agenthacker (pip install agenthacker); the import name is firewall_sdk (from firewall_sdk import Firewall). requests is installed automatically; ML layers are optional extras.

# Base install from PyPI — scan engine + high-level Firewall cloud client
pip install agenthacker

# With LLM Guard semantic scanning (DeBERTa models, ~500MB RAM per worker)
pip install "agenthacker[llm_guard]"

# With Invariant trace-level policy analysis
pip install "agenthacker[invariant]"

# From source (this repo), for local development
pip install -e .

# Dev extras (pytest, pytest-asyncio)
pip install -e ".[dev]"

For a complete runnable version, see examples/quickstart.py — python examples/quickstart.py prints a blocked and an allowed verdict with no key and no network.

The Four Checkpoints

CP-1: Input Scan

from firewall_sdk import scan_input

scan_input(text, *, max_input_length) -> ScanResult

When: Before the user's message enters the agent loop.

What it checks: R-01 through R-15: regex-based injection detection (system prompt extraction, role hijacking, hypothetical jailbreak, indirect injection, delimiter injection, RAG injection, privilege escalation, encoded content, context overflow, social engineering, adversarial suffixes). Topic scoping (R-16) is enforced at the intent gate (CP-2), not here.

Agent supplies: An integer max input length (R-09 overflow). Domain topic scoping is configured at the intent gate, not passed to scan_input.

CP-2: Data Field Scan

from firewall_sdk import scan_data_field

scan_data_field(text) -> ScanResult

When: On each field of tool results before they reach the model.

What it checks: Same injection rules as CP-1 adapted for data context (R-05 blocks any URL in data fields, not just URLs with action words), plus S-03 secrets detection.

Agent supplies: Nothing. Fully domain-agnostic.

CP-3: Tool Call Authorization

from firewall_sdk import scan_tool_call

scan_tool_call(name, args, allowed_ids, *, allowed_tools, id_resolver) -> ScanResult

When: After the LLM emits a tool_use, before execution.

What it checks: Tool name against the allowlist, then entity ID authorization via the resolver.

Agent supplies:

• allowed_tools — a request-scoped set[str] of permitted tool names. This must be computed per-request, not a static constant. The patient agent computes it from auth context to enforce role-based restrictions (e.g., patients cannot use send_reminder). The order agent passes a static {"get_order_status"}.
• id_resolver — a callable with signature (name: str, args: dict) -> set[str] that returns all entity IDs requiring authorization for this tool call. Returns an empty set if no IDs to check.

CP-4: Output Scan

from firewall_sdk import scan_output

scan_output(
    text, system_prompt, requester_email, allowed_ids, all_user_emails,
    *, entity_pattern, leakage_label
) -> ScanResult

When: After the model produces a final text response, before returning to the user.

What it checks: S-01 system prompt leakage (5-word shingle overlap), S-02 cross-entity ID and email leakage, S-03 secrets, S-04 offensive content.

Agent supplies: A compiled regex for entity IDs (e.g., re.compile(r"ORD-\d+") or re.compile(r"PAT-\d+")), and a label string for audit logs (e.g., "Cross-User Data Leakage").

ScanResult

All scan functions return a ScanResult:

from firewall_sdk import ScanResult, CLEAN

@dataclass
class ScanResult:
    clean: bool
    rule_id: str | None = None
    rule_name: str | None = None
    matched_text: str | None = None  # truncated to 100 chars

CLEAN is the singleton ScanResult(clean=True). Any non-clean result blocks the request — the agent returns a refusal and logs an audit event.

Wiring Example

See apps/order_agent/firewall.py (~70 lines) for the canonical wiring example — a real, tested thin wrapper showing exactly how an agent feeds domain config into SDK scan functions.

Optional Integrations

LLM Guard (firewall_sdk.llm_guard)

Semantic second-pass scanning using DeBERTa models. Detects prompt injection, secrets, and sensitive output that regex rules miss.

import firewall_sdk.llm_guard as llm_guard

# At startup
llm_guard.warmup(
    enabled=True,
    injection_threshold=0.9,
    use_onnx=False,
    workers=1,
)

# Per-checkpoint (same signatures as core scan functions)
await llm_guard.scan_input(text)
await llm_guard.scan_data_field(text)
await llm_guard.scan_output(prompt, output_text)

llm_guard.is_ready()  # For /health endpoint

Invariant Analyzer (firewall_sdk.invariant)

Trace-level policy evaluation using Invariant policy files (.inv). Detects sequence-level violations (retry loops, data leakage after denied tool calls, unauthorized tool calls) that per-message scanning cannot catch.

import firewall_sdk.invariant as invariant

# At startup
await invariant.warmup(
    enabled=True,
    policy_dir="policies/order_agent",
)

# After CP-4 (or CP-3 for pre-tool checks)
result = await invariant.analyze_trace(trace)

invariant.BLOCKING_RULES  # frozenset of rule names eligible for blocking mode
invariant.is_ready()      # For /health endpoint

Structured Logger (firewall_sdk.logger)

JSON-formatted audit logging with HMAC-hashed email identifiers.

import firewall_sdk.logger as logger

# At startup
logger.setup_logging("INFO", "your-secret-salt")  # Not "change-me-in-prod"

logger.log_firewall_event(checkpoint, scan_result, user_email, ip)
logger.log_agent_invocation(user_email, question, tool_calls, tokens, latency, blocked)
logger.hash_email("user@example.com")  # -> 16-char HMAC hex

Trace Normalization (firewall_sdk.trace)

Converts Anthropic-format traces to OpenAI-compatible format for Invariant analysis. Pure function, no state.

from firewall_sdk.trace import normalize_trace

invariant_trace = normalize_trace(anthropic_snapshot_trace)

Fail-Open Behavior

LLM Guard and Invariant return CLEAN when disabled, not initialized, or on error. This is a deliberate availability choice — a scanner failure does not block user requests. The /health endpoint exposes is_ready() for monitoring.

Module-Level State

All three stateful modules (llm_guard, invariant, logger) use module-level globals set by warmup(). This is safe when each agent runs in its own process. Each module provides a reset() function for test isolation (see tests/conftest.py).

Provisional Module: agent_helpers

from firewall_sdk.agent_helpers import (
    needs_llm_guard,       # Domain-agnostic, likely to stabilize
    normalize_refusal,     # Domain-agnostic, likely to stabilize
    serialize_blocks,      # Anthropic-specific, will change in Phase 5
    serialize_message,     # Anthropic-specific, will change in Phase 5
    snapshot_trace,        # Anthropic-specific, will change in Phase 5
)

Both agents use all five functions. They work. But the interface is not frozen.

serialize_blocks, serialize_message, and snapshot_trace access Anthropic SDK content block attributes directly (.type, .text, .id, .name, .input). These will change when framework portability is introduced. needs_llm_guard and normalize_refusal are domain-agnostic and likely to stabilize.

agent_helpers is not re-exported from firewall_sdk.__init__. Import it directly: from firewall_sdk.agent_helpers import ....

What the Agent Must Supply

Each agent provides domain config through a thin firewall.py wrapper that imports SDK functions and passes domain-specific values. The agent's agent.py calls the wrapper identically to pre-extraction code.

See apps/order_agent/firewall.py as the canonical wiring example.

What the SDK Does NOT Own

The SDK must never contain:

• Agent loop or orchestration logic
• FastAPI routes or HTTP request/response schemas
• Auth implementations or identity provider logic
• Storage adapters or data access code
• Domain prompts or domain-specific policy rules
• Tool implementations
• UI code

If it mentions a domain noun (orders, patients, appointments), it does not belong in the SDK.

Initialization Lifecycle

startup:
  logger.setup_logging(level, salt)
  llm_guard.warmup(enabled=..., threshold=..., onnx=..., workers=...)
  await invariant.warmup(enabled=..., policy_dir=...)

per-request:
  allowed_tools = compute from auth context (request-scoped)
  CP-1: firewall.scan_input -> optionally llm_guard.scan_input
  agent loop:
    CP-3: firewall.scan_tool_call(..., allowed_tools=allowed_tools)
    execute tool
    CP-2: firewall.scan_data_field per field -> optionally llm_guard.scan_data_field
  CP-4: firewall.scan_output -> optionally llm_guard.scan_output
        -> optionally invariant.analyze_trace

test teardown:
  llm_guard.reset(); invariant.reset(); logger.reset()

Architecture

Checkpoint Sequence

sequenceDiagram
    participant U as User
    participant A as Agent
    participant F as Firewall SDK
    participant T as Tool
    participant M as Model (Claude)

    U->>A: Request
    A->>F: CP-1: scan_input
    alt blocked
        F-->>A: ScanResult(clean=False)
        A-->>U: Refusal + audit log
    end
    A->>M: System prompt + tools + message
    M->>A: tool_use
    A->>F: CP-3: scan_tool_call
    alt blocked
        F-->>A: ScanResult(clean=False)
        A->>M: "Access denied" tool result
    else clean
        A->>T: Execute tool
        T->>A: Result
        A->>F: CP-2: scan_data_field (per field)
        alt blocked
            A->>M: "Data unavailable" tool result
        else clean
            A->>M: Tool result
        end
    end
    M->>A: end_turn (final text)
    A->>F: CP-4: scan_output
    alt blocked
        F-->>A: ScanResult(clean=False)
        A-->>U: Refusal + audit log
    else clean
        A-->>U: Response
    end

SDK Composition

graph TB
    subgraph "Core (stable, re-exported from __init__)"
        schemas["schemas<br/>ScanResult, CLEAN"]
        scan_engine["scan_engine<br/>scan_input, scan_data_field"]
        tool_guard["tool_guard<br/>scan_tool_call"]
        output_guard["output_guard<br/>scan_output"]
    end

    subgraph "Optional integrations (stable, import by path)"
        llm_guard["llm_guard<br/>warmup, scan_*, reset"]
        invariant["invariant<br/>warmup, analyze_trace, reset"]
        sdk_logger["logger<br/>setup_logging, log_*, reset"]
        trace["trace<br/>normalize_trace"]
    end

    subgraph "Provisional (will change in Phase 5)"
        helpers["agent_helpers<br/>serialize_*, normalize_refusal, needs_llm_guard"]
    end

    OA[Order Agent] --> schemas & scan_engine & tool_guard & output_guard
    OA --> llm_guard & invariant & sdk_logger
    OA --> helpers

    PA[Patient Agent] --> schemas & scan_engine & tool_guard & output_guard
    PA --> llm_guard & invariant & sdk_logger
    PA --> helpers

Version and Governance

Current version: 0.1.0 (semver 0.y.z = initial development).

1. No breaking changes to firewall_sdk without updating both agents in the same PR
2. Public interface modifications require CHANGELOG entries
3. __all__ lists and tests/firewall_sdk/test_public_api_surface.py enforce what's public

Known Couplings and Future Work

Anthropic-specific serialization. agent_helpers.serialize_blocks, serialize_message, and snapshot_trace access Anthropic SDK content block attributes. trace.normalize_trace converts Anthropic format to OpenAI-compatible format for Invariant. Phase 5 must abstract these for framework portability. This is why agent_helpers is provisional. (trace is semi-public because its interface is stable — one pure function — even though its implementation is Anthropic-format-aware.)

Module-level state. The warmup()/reset() pattern is process-scoped. Multi-agent-in-one-process hosting would require instance-based refactoring.

_LLM_REFUSAL_INDICATORS duplication. Both agents define nearly identical refusal regexes — one says "specifically designed to help you check", the other says "specifically designed to help". Could become an SDK-provided default with agent override in a future version.