API Documentation
Everything you need to integrate Snapwire with your AI agents. Intercept tool calls, enforce rules, and keep humans in the loop.
Authentication
All API requests require an API key passed via the Authorization header using Bearer token format.
HTTP
Authorization: Bearer af_your_api_key_here
Generate API keys from the dashboard under the API Keys tab. Each key is scoped to a specific workspace (personal or organization).
Quick Start
Here's the simplest integration — send a tool call to Snapwire before executing it:
Python
import requests
API_KEY = "af_your_key_here"
BASE_URL = "https://app.getsnapwire.com"
def check_tool_call(tool_name, parameters, intent=""):
response = requests.post(
f"{BASE_URL}/api/intercept",
headers={"Authorization": f"Bearer {API_KEY}"},
json={
"tool_name": tool_name,
"parameters": parameters,
"intent": intent,
"agent_id": "my-agent"
}
)
result = response.json()
if result["status"] == "allowed":
return True # Safe to execute
elif result["status"] == "pending":
return None # Held for human review
else:
return False # Blocked by rules
# Example: check before sending an email
allowed = check_tool_call(
tool_name="send_email",
parameters={"to": "user@example.com", "subject": "Report"},
intent="Send weekly analytics report"
)
if allowed:
send_email(...) # Your actual tool call
Try It Playground
Test a live tool call against your Snapwire instance. This sends a real request to POST /api/intercept so you can see how Snapwire evaluates and responds.
Try something malicious — Snapwire will catch it.
Allowed
The tool call passed all rules and is safe to execute. No human review needed.
Blocked
The tool call violated one or more rules and was denied. The response includes the violations and reasons.
Pending
The tool call was flagged for human review. A Snap-Card is created in the dashboard for approve/deny/edit.
Intercept Tool Call
The primary endpoint. Send every tool call here before execution. Snapwire evaluates it against your constitutional rules and returns a decision.
Request Body
| Parameter | Type | Description |
|---|---|---|
| tool_name REQUIRED | string | Name of the tool being called (e.g., "send_email", "delete_file") |
| parameters | object | The parameters being passed to the tool |
| intent | string | A description of why the agent wants to call this tool |
| context | string | Additional context about the agent's current task |
| agent_id | string | Identifier for the agent making the call |
| parent_agent_id | string | ID of the parent agent that spawned this agent (for multi-agent chain traceability) |
| inner_monologue | string | Agent's internal reasoning (used by heuristic goal-drift analysis) |
| webhook_url | string | URL to receive a callback when a pending action is resolved |
| usage | object | Token usage from the LLM response. Fields: thinking_tokens, input_tokens, output_tokens. Used by the Thinking Token Sentinel. |
JSON
{
"status": "allowed",
"action_id": "a1b2c3d4",
"risk_score": 15,
"analysis": "Action is within policy bounds."
}
JSON
{
"status": "pending",
"action_id": "e5f6g7h8",
"risk_score": 78,
"violations": [
{
"rule": "no_external_emails",
"severity": "high",
"reason": "Sending email to external address requires approval."
}
],
"message": "Action held for human review."
}
Get Pending Actions
Retrieve all actions currently waiting for human review. Useful for building custom review UIs or monitoring queues.
Resolve an Action
Approve or deny a pending action programmatically.
| Parameter | Type | Description |
|---|---|---|
| decision REQUIRED | string | "approve" or "deny" |
Audit Log
Retrieve the complete audit trail of all intercepted actions. Supports pagination with ?limit=50&offset=0 query parameters.
Constitution Rules
List all active constitutional rules for the current workspace.
Create a new constitutional rule.
| Parameter | Type | Description |
|---|---|---|
| rule_name REQUIRED | string | Unique identifier for the rule |
| value REQUIRED | string | The rule text in plain English |
| severity | string | "low", "medium", "high", or "critical" |
| mode | string | "enforce" (block violations) or "monitor" (log only) |
API Keys
List all API keys for the current workspace.
Generate a new API key.
Webhooks
Register a webhook to receive real-time notifications when actions are resolved. Useful for automated pipelines.
Tool Safety Catalog
Retrieve the safety catalog showing all known tools, their safety grades (A–F), and approval status.
Honeypot Tripwires
Manage decoy tools. When an agent calls a honeypot, the system auto-locks the API key and creates an alert.
Identity Vault
Manage credential entries. The vault injects credentials into approved tool calls so agents never see raw secrets. Agents receive Snap-Tokens instead of real credentials.
Loop Detection (Fuse Breaker)
Snapwire's Loop Detector automatically identifies when an agent is stuck in a repetitive cycle — calling the same tool with the same parameters over and over. When a loop is detected, Snapwire triggers a Fuse Breaker that blocks further calls from that agent until the loop clears, preventing runaway behavior and wasted resources.
How Fuse Breaker Works
The detector tracks a rolling window of recent tool calls per agent. If the same tool_name + parameters combination appears more than a configurable threshold (default: 5 times in 60 seconds), the agent is flagged. Subsequent calls are blocked with a 429 status and a blocked-loop audit entry until the window expires.
Get Loop Events
Retrieve recent loop detection events for your workspace.
| Parameter | Type | Description |
|---|---|---|
| limit | integer | Number of events to return (default: 20) |
JSON
{
"events": [
{
"id": 1,
"agent_id": "research-agent",
"tool_name": "web_search",
"repeat_count": 7,
"pattern_hash": "a1b2c3d4",
"detected_at": "2026-02-24T10:30:00Z",
"window_seconds": 60
}
]
}
Get Loop Statistics
Get aggregate loop detection statistics for your workspace, including total loops detected and most common offending agents.
JSON
{
"total_loops_detected": 42,
"unique_agents": 3,
"top_agents": [
{"agent_id": "research-agent", "loop_count": 25},
{"agent_id": "data-agent", "loop_count": 12}
],
"top_tools": [
{"tool_name": "web_search", "loop_count": 30}
]
}
Observe & Audit Mode
Observe & Audit Mode lets you observe what Snapwire would block without actually blocking anything. All tool calls pass through, but violations are logged as shadow-blocked in the audit log. This is ideal for onboarding new rules, testing configurations, or monitoring agents in production without disrupting their work.
Modes
Observe & Audit — All tool calls are allowed. Violations are logged but not enforced. Responses include "shadow_mode": true.
Enforce (block) — Violations are actively blocked. This is the default production mode.
Disabled — No rule evaluation occurs (not recommended).
Get Observe & Audit Mode Status
JSON
{
"shadow_mode": true,
"changed_at": "2026-02-24T09:15:00Z",
"changed_by": 1
}
Toggle Observe & Audit Mode
| Parameter | Type | Description |
|---|---|---|
| enabled | boolean | Set to true for observe & audit mode, false for enforce mode. Omit to toggle. |
JSON
{
"shadow_mode": false,
"message": "Blocking Mode enabled - violations will be blocked"
}
Trust Rules
Trust Rules let you auto-approve specific tool + agent combinations so trusted, routine actions don't require human review. When you approve a pending action, you can create a Trust Rule that auto-approves identical future calls for a set duration (default: 24 hours). This dramatically reduces review queue noise for known-safe patterns.
How 24-Hour Auto-Approve Works
When a reviewer approves a pending action, they can check "Trust this pattern." Snapwire creates a Trust Rule matching the tool_name and agent_id. For the next 24 hours, matching calls skip the review queue and are auto-approved. Trust Rules expire automatically, and admins can revoke them early if needed.
List Trust Rules
Retrieve all active (non-expired) trust rules for your workspace.
JSON
{
"trust_rules": [
{
"id": 1,
"tool_name": "send_email",
"agent_id": "report-agent",
"created_at": "2026-02-24T08:00:00Z",
"expires_at": "2026-02-25T08:00:00Z",
"is_active": true,
"approved_by": 1
}
]
}
Revoke a Trust Rule
Immediately deactivate a trust rule. Requires admin access. Future matching calls will go back to the review queue.
JSON
{
"status": "revoked",
"rule": {
"id": 1,
"tool_name": "send_email",
"agent_id": "report-agent",
"is_active": false
}
}
Snap-Tokens (Credential Proxy)
Snap-Tokens are proxy tokens that replace real credentials in agent workflows. Instead of giving your agent a raw API key or database password, you store the credential in the Identity Vault and issue a Snap-Token. The agent uses the Snap-Token, and Snapwire injects the real credential only into approved tool calls. If an agent is compromised, revoke the Snap-Token — the real credential is never exposed.
Credential Proxy Flow
1. Store a credential in the Identity Vault.
2. Generate a Snap-Token linked to that vault entry.
3. Give the Snap-Token to your agent (e.g., snap_abc123).
4. The agent passes "proxy_token": "snap_abc123" in the intercept request.
5. If the call is approved, Snapwire resolves the token and injects the real credential into the response.
List Proxy Tokens
List all proxy tokens for your workspace.
JSON
{
"tokens": [
{
"id": 1,
"token_prefix": "snap_abc1...",
"vault_entry_id": 5,
"label": "Stripe key for billing-agent",
"created_at": "2026-02-24T10:00:00Z",
"is_active": true
}
]
}
Create a Proxy Token
| Parameter | Type | Description |
|---|---|---|
| vault_entry_id REQUIRED | integer | ID of the vault entry to link this token to |
| label | string | Human-readable label for the token (e.g., "Stripe key for billing-agent") |
JSON
{
"token": "snap_a1b2c3d4e5f6",
"token_id": 1,
"vault_entry_id": 5,
"label": "Stripe key for billing-agent"
}
Important: The full token value is only returned once at creation. Store it securely.
Revoke a Proxy Token
Revoke a single proxy token. The linked credential remains in the vault.
Revoke All Proxy Tokens
Revoke all proxy tokens in the workspace. Use this for emergency credential rotation.
JSON
{
"status": "all_revoked",
"count": 5
}
Config Export / Import
Export your entire rule configuration as a portable JSON file, and import it into another Snapwire instance. This is useful for migrating rules between environments (dev → staging → prod), sharing configurations across teams, or backing up your setup.
Export Rules
Export all rules as a JSON object.
JSON
{
"_meta": {
"generator": "Snapwire",
"version": "1.0.0",
"exported_at": "2026-02-24T12:00:00Z",
"rule_count": 8
},
"rules": [
{
"name": "no_external_emails",
"description": "Block emails to external domains",
"rule_text": "Do not allow send_email to non-company domains",
"severity": "high",
"category": "Email Safety",
"enabled": true
}
]
}
Download Rules as File
Download the export as a .json file attachment (Content-Disposition header set). File is named snapwire-rules-YYYY-MM-DD.json.
Import Rules
Import rules from a Snapwire export file. Accepts either a JSON body or a multipart file upload. Requires admin access. Existing rules with the same name are skipped (not overwritten).
| Parameter | Type | Description |
|---|---|---|
| _meta REQUIRED | object | Must contain "generator": "Snapwire" |
| rules REQUIRED | array | Array of rule objects (see export format above) |
JSON
{
"status": "imported",
"imported_count": 6,
"skipped_count": 2,
"skipped_rules": ["no_external_emails", "budget_cap"],
"message": "Successfully imported 6 rule(s). Skipped 2 existing rule(s)."
}
cURL (file upload)
curl -X POST https://app.getsnapwire.com/api/rules/import \
-H "Authorization: Bearer af_your_key_here" \
-F "file=@snapwire-rules-2026-02-24.json"
Blast Radius Governor
The Blast Radius Governor prevents runaway agents from overwhelming your systems. It enforces per-agent rate limits on tool calls within a rolling time window. When an agent exceeds its limit, it is locked out for a configurable duration.
Get Configuration
Retrieve the current blast radius settings for your workspace.
JSON
{
"enabled": true,
"max_calls": 50,
"window_seconds": 300,
"lockout_seconds": 600,
"max_spend_per_session": 10.0,
"require_manual_reset": false
}
Update Configuration
Update blast radius settings. Requires admin access. All fields are optional — only include what you want to change.
| Parameter | Type | Description |
|---|---|---|
| enabled | boolean | Enable or disable the governor |
| max_calls | integer | Maximum tool calls per agent per window |
| window_seconds | integer | Rolling window duration in seconds |
| lockout_seconds | integer | How long an agent is locked out after exceeding the limit |
| max_spend_per_session | float | Maximum estimated cost per agent session |
| require_manual_reset | boolean | If true, locked-out agents require manual admin clearance |
Get Blast Radius Events
List recent blast radius events (agents that hit or approached their limits).
Get Active Lockouts
List agents currently locked out by the blast radius governor.
Clear a Lockout
Manually clear a lockout for a specific agent. Requires admin access.
JSON
{
"status": "cleared",
"agent_id": "research-agent"
}
Rate Limits
Snapwire enforces per-API-key rate limits on the /api/intercept endpoint to protect against abuse and ensure fair usage. The default limit is 60 requests per minute per API key.
How It Works
Each API key has an independent counter that resets every 60 seconds. When the limit is reached, Snapwire returns a 429 Too Many Requests response. The response includes a Retry-After-style field indicating when the window resets.
Every successful intercept response includes rate limit metadata:
JSON (in intercept response)
{
"status": "allowed",
"rate_limit": {
"remaining": 45,
"reset_at": 1708790460
}
}
Check Rate Limit Status
Get rate limit usage for all API keys in your workspace. Requires admin access.
JSON
{
"global_limit": 60,
"keys": [
{
"key_id": 1,
"key_name": "Production Key",
"agent_name": "research-agent",
"limit": 60,
"request_count": 15,
"requests_remaining": 45,
"reset_at": 1708790460
}
]
}
Update Global Rate Limit
Change the global rate limit for all API keys. Requires admin access.
| Parameter | Type | Description |
|---|---|---|
| limit REQUIRED | integer | New rate limit per minute (1–1000) |
JSON
{
"status": "updated",
"global_limit": 120
}
Handling Rate Limit Errors
When rate limited, implement exponential backoff. Check the reset_at field (Unix timestamp) to know exactly when to retry:
Python
import time, requests
def intercept_with_retry(tool_name, params, max_retries=3):
for attempt in range(max_retries):
resp = requests.post(
f"{BASE_URL}/api/intercept",
headers={"Authorization": f"Bearer {API_KEY}"},
json={"tool_name": tool_name, "parameters": params}
)
if resp.status_code != 429:
return resp.json()
reset_at = resp.json().get("rate_limit", {}).get("reset_at", 0)
wait = max(1, reset_at - time.time())
time.sleep(wait)
raise Exception("Rate limit exceeded after retries")
Error Reference
All Snapwire API errors return a JSON body with an "error" field describing the issue. Here are the HTTP status codes you may encounter:
| Status Code | Meaning | Common Causes | How to Handle |
|---|---|---|---|
| 400 Bad Request | Invalid or missing request data | Missing tool_name, invalid JSON, malformed parameters | Check your request body matches the documented schema |
| 401 Unauthorized | Authentication failed | Missing or invalid API key, expired session | Verify your Authorization: Bearer header contains a valid, active API key |
| 403 Forbidden | Action blocked by policy | Rule violation, honeypot triggered, deception detected, tool blocked in catalog, input sanitizer flagged malicious content, admin-only endpoint | Check the violations array for specific rule details; review your constitutional rules |
| 412 Precondition Required | Reasoning required for high-risk action | Tool call has critical or high severity violations but no inner_monologue was provided | Re-submit the request with the inner_monologue field populated, explaining why the action is needed. The action will then be evaluated normally. |
| 404 Not Found | Resource does not exist | Invalid action ID, rule ID, or vault entry ID | Verify the resource ID is correct and belongs to your workspace |
| 429 Too Many Requests | Rate limit or blast radius exceeded | Too many API calls per minute, agent locked out by blast radius governor, agent stuck in a loop | Check rate_limit.reset_at or blast_radius fields; implement exponential backoff |
| 500 Internal Server Error | Unexpected server error | Database issues, AI auditor unavailable | Retry after a short delay; if persistent, check the /health endpoint and server logs |
Error Response Format
JSON (403 example)
{
"status": "blocked",
"message": "Tool call blocked: violates constitutional rule.",
"violations": [
{
"rule": "no_external_emails",
"severity": "high",
"reason": "Sending email to external address requires approval."
}
]
}
JSON (429 example)
{
"error": "Rate limit exceeded. Please slow down.",
"rate_limit": {
"remaining": 0,
"reset_at": 1708790460
}
}
JSON (412 example — Reasoning Required)
{
"status": "reasoning_required",
"message": "High-risk action detected. Re-submit this request with the 'inner_monologue' field populated, explaining why this action is needed.",
"violations": [
{
"rule": "no_destructive_ops",
"severity": "critical",
"reason": "Deleting production database requires explicit justification."
}
],
"risk_score": 85
}
JSON (401 example)
{
"error": "Authentication required. Provide an API key via Authorization header or sign in."
}
MCP Compatibility
Snapwire is a Deterministic Gateway for MCP-Compliant Agents. The /api/intercept endpoint natively accepts the Model Context Protocol (MCP) JSON-RPC 2.0 format. If you're running an MCP-compatible agent, you can send requests directly without any translation layer.
How It Works
Snapwire auto-detects MCP requests by checking for jsonrpc and method fields. When detected, it translates the MCP format to Snapwire's native format, runs all security checks, and wraps the response back in JSON-RPC 2.0.
Supported MCP Method
| Method | Description |
|---|---|
tools/call | Intercept and audit a tool call before execution |
Example Request
JSON (MCP)
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "send_email",
"arguments": {
"to": "user@example.com",
"subject": "Weekly Report"
}
}
}
JSON
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"status": "allowed",
"audit": { ... },
"message": "Tool call passed all constitutional checks."
}
}
JSON
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32603,
"message": "Tool call blocked. Awaiting manual approval.",
"data": {
"status": "blocked",
"action_id": "a1b2c3d4",
"audit": { ... }
}
}
}
MCP format detection is automatic — the same /api/intercept endpoint handles both Snapwire-native and MCP JSON-RPC formats. No configuration needed.
Thinking Token Sentinel
The Thinking Token Sentinel monitors extended thinking usage from models like Claude 3.7+ that expose thinking_tokens in their usage metadata. If an agent's model consumes more than 50,000 thinking tokens without producing a tool call, Snapwire fires a "Potential Logic Loop" advisory warning.
How to Report Usage
Include the usage object in your intercept request with token counts from your LLM provider's response:
JSON
{
"tool_name": "analyze_data",
"parameters": { "query": "Q4 revenue" },
"usage": {
"thinking_tokens": 62000,
"input_tokens": 1500,
"output_tokens": 800
}
}
Sentinel Response
When triggered, the response includes a thinking_sentinel_warning field (the tool call still proceeds through the normal pipeline — this is advisory, not a block):
JSON
{
"status": "allowed",
"thinking_sentinel_warning": {
"triggered": true,
"severity": "warning",
"thinking_tokens": 62000,
"threshold": 50000,
"estimated_cost": 0.62,
"message": "Potential Logic Loop: 62,000 thinking tokens consumed without a tool call."
}
}
Note: The Thinking Token Sentinel is a heuristic cost-management tool based on token-to-action ratios. It is not a guarantee of model accuracy or behavior.
Sentinel Proxy
The Sentinel Proxy is a transparent reverse proxy that intercepts LLM API traffic and routes it through Snapwire's governance pipeline. No SDK integration or code changes required — just swap your API base URL.
Quick Start
Shell
# Docker (recommended — starts Snapwire + Sentinel + PostgreSQL):
docker-compose up
# Standalone:
python -m sentinel
# Point your agent at the proxy:
OPENAI_BASE_URL=http://localhost:8080/v1 python my_agent.py
Supported Protocols
The Sentinel Proxy automatically detects tool-call patterns across multiple protocols:
| Protocol | Detection Pattern | Confidence |
|---|---|---|
| OpenAI | tools array, tool_choice, tool_calls in messages | 0.90-0.95 |
| Anthropic | tool_use content blocks, tools definitions | 0.90-0.95 |
| Google Gemini | function_declarations, functionCall in parts/candidates | 0.90-0.95 |
| Cohere | parameter_definitions, tool_calls with parameters | 0.85-0.95 |
| AWS Bedrock | toolConfig.tools[].toolSpec, toolUse content blocks | 0.90-0.95 |
| LangChain | tool_calls with args dict, ToolMessage type | 0.80-0.95 |
| MCP (JSON-RPC) | jsonrpc: "2.0" + method: "tools/call" | 1.00 |
| A2A (Agent-to-Agent) | tasks/send, delegate_task | 0.85 |
| Generic JSON-RPC | Any method + params pattern | 0.50-0.80 |
Also compatible via OpenAI format: Mistral, Azure OpenAI, and any OpenAI-compatible provider are automatically detected by the OpenAI detector since they use the same wire format.
Custom Detectors & Detector Lab
Need to support a provider not listed above? Snapwire's Detector Lab (Engine Room → Detector Lab) lets you paste a sample JSON payload from any LLM provider and generates a working detector function automatically. One click to save and activate — no code editing required.
Custom detectors are stored in sentinel/custom_detectors.py and auto-loaded at startup. This file survives upstream updates, so your custom protocol support persists when you pull new versions of Snapwire.
Developers can also write detectors manually using the @register_protocol decorator:
Python
from sentinel.detector import register_protocol, DetectedToolCall
@register_protocol
def detect_my_provider(body: dict) -> list:
results = []
tool_calls = body.get("my_tool_calls")
if isinstance(tool_calls, list):
for tc in tool_calls:
if isinstance(tc, dict) and "name" in tc:
results.append(DetectedToolCall(
tool_name=tc["name"],
parameters=tc.get("args", {}),
protocol="my_provider",
confidence=0.9
))
return results
Operational Modes
| Mode | Logging | Headers | Blocking | Fail Behavior |
|---|---|---|---|---|
observe | Yes (non-blocking) | No | No | N/A (zero latency impact) |
audit | Yes | Yes | No | Pass through |
enforce | Yes | Yes | Yes | Fail-closed (block if Snapwire unreachable) |
Identity Headers
In audit and enforce modes, the proxy injects NIST-compliant identity headers:
| Header | Purpose | NIST Alignment |
|---|---|---|
X-Snapwire-Origin-ID | Root human/principal identity | Multimodal Identity Lineage (RFI 2a.iii) |
X-Snapwire-Authorized-By | Accountable human principal who authorized this agent | Human Accountability (Colorado SB24-205) |
X-Snapwire-Parent-ID | Immediate caller in delegation chain | Delegation Trace (NISTIR 8596) |
X-Snapwire-Trace | Unique per-request trace ID | Forensic Non-Repudiation |
X-Snapwire-Signature | HMAC-SHA256 proof of governance | Cryptographic Non-Repudiation |
X-Snapwire-Timestamp | Unix timestamp for replay protection | Temporal Binding |
The X-Snapwire-Signature header is an HMAC-SHA256 digest of trace_id.timestamp.request_path, signed with your SNAPWIRE_SIGNING_SECRET. Downstream services can verify this signature to cryptographically prove the request passed through Snapwire governance.
Configuration
| Environment Variable | Default | Description |
|---|---|---|
SENTINEL_PORT | 8080 | Port the proxy listens on |
SENTINEL_MODE | audit | Operational mode: observe, audit, enforce |
UPSTREAM_URL | https://api.openai.com | Real LLM API endpoint |
SNAPWIRE_URL | http://localhost:5000 | Snapwire governance server URL |
SNAPWIRE_API_KEY | (empty) | API key for Snapwire authentication |
SENTINEL_AGENT_ID | sentinel-proxy | Default agent ID for lineage tracking |
SENTINEL_ORIGIN_ID | human-principal | Root identity for NIST lineage chain |
SENTINEL_AUTHORIZED_BY | (defaults to SENTINEL_ORIGIN_ID) | Accountable human principal for X-Snapwire-Authorized-By header |
SNAPWIRE_SIGNING_SECRET | (empty) | HMAC-SHA256 secret for X-Snapwire-Signature header |
NIST Compliance Export
Shell
# Generate NIST compliance mapping report:
python -m sentinel --export-nist
# Output: snapwire-nist-mapping.md
# Includes: compliance scorecard (A/B/C grade), NIST RFI mapping,
# CSF 2.0 category coverage, active rules summary, legal framework
The export maps your current configuration to NIST RFI requirements (Docket NIST-2025-0035) and includes a compliance grade based on your operational mode.
Python SDK
A minimal Python wrapper for quick integration:
Python
import requests
class AgenticFirewall:
def __init__(self, api_key, base_url="https://app.getsnapwire.com"):
self.api_key = api_key
self.base_url = base_url
self.headers = {"Authorization": f"Bearer {api_key}"}
def check(self, tool_name, parameters=None, intent="", agent_id="default", usage=None):
payload = {
"tool_name": tool_name,
"parameters": parameters or {},
"intent": intent,
"agent_id": agent_id
}
if usage:
payload["usage"] = usage
resp = requests.post(
f"{self.base_url}/api/intercept",
headers=self.headers,
json=payload
)
return resp.json()
def get_pending(self):
return requests.get(
f"{self.base_url}/api/actions/pending",
headers=self.headers
).json()
def resolve(self, action_id, decision):
return requests.post(
f"{self.base_url}/api/actions/{action_id}/resolve",
headers=self.headers,
json={"decision": decision}
).json()
# Usage
fw = AgenticFirewall("af_your_key_here")
result = fw.check("send_email", {"to": "user@example.com"}, "Send report")
print(result["status"]) # "allowed", "pending", or "blocked"
Node.js SDK
JavaScript
class AgenticFirewall {
constructor(apiKey, baseUrl = "https://app.getsnapwire.com") {
this.apiKey = apiKey;
this.baseUrl = baseUrl;
}
async check(toolName, parameters = {}, intent = "", agentId = "default") {
const resp = await fetch(`${this.baseUrl}/api/intercept`, {
method: "POST",
headers: {
"Authorization": `Bearer ${this.apiKey}`,
"Content-Type": "application/json"
},
body: JSON.stringify({
tool_name: toolName,
parameters,
intent,
agent_id: agentId
})
});
return resp.json();
}
async resolve(actionId, decision) {
const resp = await fetch(`${this.baseUrl}/api/actions/${actionId}/resolve`, {
method: "POST",
headers: {
"Authorization": `Bearer ${this.apiKey}`,
"Content-Type": "application/json"
},
body: JSON.stringify({ decision })
});
return resp.json();
}
}
// Usage
const fw = new AgenticFirewall("af_your_key_here");
const result = await fw.check("delete_file", { path: "/data/old.csv" });
console.log(result.status); // "allowed", "pending", or "blocked"
Framework Integrations
Pre-built integration examples for popular AI frameworks. Each example shows how to wrap Snapwire into your existing agent pipeline.
LangChain
Wrap any LangChain tool with Snapwire protection. Blocked actions return a safe message instead of executing.
Python
from langchain.tools import Tool
from agentic_firewall import AgenticFirewall
fw = AgenticFirewall(api_key="YOUR_API_KEY", url="https://your-server.com")
class FirewallWrappedTool(Tool):
def _run(self, query: str) -> str:
result = fw.intercept(
tool_name=self.name,
parameters={"query": query},
agent_id="langchain-agent"
)
if result["decision"] == "blocked":
return f"Action blocked: {result['reason']}"
return self.original_func(query)
CrewAI
Use a before-tool callback to intercept every tool call made by CrewAI agents. Blocked actions raise an exception to halt execution.
Python
from crewai import Agent, Task, Crew
from agentic_firewall import AgenticFirewall
fw = AgenticFirewall(api_key="YOUR_API_KEY", url="https://your-server.com")
def firewall_callback(tool_name, params):
result = fw.intercept(
tool_name=tool_name,
parameters=params,
agent_id="crewai-agent"
)
if result["decision"] == "blocked":
raise Exception(f"Blocked by firewall: {result['reason']}")
agent = Agent(
role="Research Analyst",
goal="Analyze data safely",
before_tool_callback=firewall_callback
)
OpenAI Assistants
Intercept function calls from OpenAI Assistants API before executing them. Returns an error payload for blocked actions.
Python
import openai
from agentic_firewall import AgenticFirewall
fw = AgenticFirewall(api_key="YOUR_API_KEY", url="https://your-server.com")
def handle_tool_call(tool_call):
result = fw.intercept(
tool_name=tool_call.function.name,
parameters=json.loads(tool_call.function.arguments),
agent_id="openai-assistant"
)
if result["decision"] == "blocked":
return {"error": f"Blocked: {result['reason']}"}
return execute_tool(tool_call)