Docs - Clampd | The Tool Call Firewall for AI Agents

Python TypeScript CLI Architecture

Python SDK

Python Quickstart

1-line integration for OpenAI, Anthropic, LangChain, Google ADK, and any function.

Guard OpenAI tool calls

 python
import clampd
from openai import OpenAI

client = clampd.openai(OpenAI(), agent_id="my-agent", secret="ags_...")
# That's it. All tool calls now go through the security pipeline.
# Or set JWT_SECRET env var - no code change needed.
response = client.chat.completions.create(model="gpt-4o", messages=messages, tools=tools)

Guard Anthropic / Claude

 python
import clampd
from anthropic import Anthropic

client = clampd.anthropic(Anthropic(), agent_id="my-agent", secret="ags_...")
# tool_use blocks are automatically intercepted and verified

Guard any function

 python
import clampd

@clampd.guard("database.query", agent_id="my-agent", secret="ags_...")
def run_query(sql: str) -> str:
    return db.execute(sql)

LangChain (callback - guards ALL tools)

 python
import clampd

agent.invoke(input, config={"callbacks": [clampd.langchain(agent_id="my-agent", secret="ags_...")]})

Google ADK

 python
import clampd

agent = Agent(model="gemini-2.0-flash", tools=[search], before_tool_callback=clampd.adk(agent_id="my-agent", secret="ags_..."))

Response Inspection (opt-in)

Add check_response=True to also inspect tool responses for PII, data anomalies, or poisoned data. Off by default - zero overhead unless enabled.

 python
# Check both requests AND responses
client = clampd.openai(OpenAI(), agent_id="my-agent", secret="ags_...", check_response=True)

@clampd.guard("database.query", agent_id="my-agent", secret="ags_...", check_response=True)
def run_query(sql: str) -> str:
    return db.execute(sql)

# Google ADK: returns (before_cb, after_cb) tuple
before_cb, after_cb = clampd.adk(agent_id="my-agent", secret="ags_...", check_response=True)
agent = Agent(before_tool_callback=before_cb, after_tool_callback=after_cb)

Multi-Agent (per-agent identity)

Each agent authenticates independently with its own ags_ secret. Kill/rate-limit/EMA operate per-agent. Delegation chains are tracked automatically.

 python
import os, clampd

clampd.init(
    agent_id="orchestrator",
    api_key=os.environ["CLAMPD_API_KEY"],
    agents={
        "orchestrator": os.environ["CLAMPD_SECRET_orchestrator"],
        "research-agent": os.environ["CLAMPD_SECRET_research_agent"],
    },
)

# research-agent gets its own JWT. Kill it from dashboard
# without affecting orchestrator.
@clampd.guard("web.search", agent_id="research-agent")
def search(query: str):
    return web_search(query)

Streaming Guard (opt-in)

Set guard_stream=True to intercept tool calls in streaming responses. Text chunks pass through immediately.

 python
client = clampd.openai(OpenAI(), agent_id="my-agent", guard_stream=True)

stream = client.chat.completions.create(model="gpt-4o", stream=True, tools=tools, messages=messages)
# Tool calls buffered → guarded → released. Text streams instantly.

CrewAI

 python
from clampd.crewai_callback import ClampdCrewAIGuard

clampd.init(agent_id="crew-agent", secret="ags_...")
guard = ClampdCrewAIGuard()
safe_tool = guard.wrap_tool(my_tool)

MCP Proxy (wrap any MCP server)

 bash
python -m clampd.mcp_server \
  --downstream "npx -y @modelcontextprotocol/server-filesystem /tmp" \
  --agent-id "your-agent-id" \
  --gateway "http://localhost:<port>"

TypeScript SDK

TypeScript Quickstart

1-line integration for OpenAI, Anthropic, and any async function.

Guard OpenAI tool calls

 typescript
import clampd from "@clampd/sdk";

const wrapped = clampd.openai(client, { agentId: "my-agent", secret: "ags_..." });
// All tool calls now verified through the security pipeline

Guard Anthropic / Claude

 typescript
import clampd from "@clampd/sdk";

const wrapped = clampd.anthropic(client, { agentId: "my-agent", secret: "ags_..." });
// tool_use blocks intercepted automatically

Guard any function

 typescript
import clampd from "@clampd/sdk";

const guarded = clampd.guard(myFn, { agentId: "my-agent", secret: "ags_...", toolName: "db.query" });
const result = await guarded({ sql: "SELECT 1" });

Wrap OpenAI tool definitions

 typescript
import clampd from "@clampd/sdk";

const guardedTools = clampd.tools(myTools, { agentId: "my-agent", secret: "ags_..." });
// Every tool.execute() now goes through the pipeline

Response Inspection (opt-in)

 typescript
import clampd from "@clampd/sdk";

// Inspect tool responses for PII, anomalies, poisoned data
const guarded = clampd.guard(myFn, { agentId: "my-agent", secret: "ags_...", toolName: "db.query", checkResponse: true });

Multi-Agent (per-agent identity)

Each agent gets its own JWT. Kill/rate-limit/EMA operate independently. Delegation chains tracked via AsyncLocalStorage.

 typescript
import clampd from "@clampd/sdk";

clampd.init({
  agentId: "orchestrator",
  apiKey: process.env.CLAMPD_API_KEY,
  agents: {
    "orchestrator": process.env.CLAMPD_SECRET_orchestrator,
    "research-agent": process.env.CLAMPD_SECRET_research_agent,
  },
});

// research-agent gets its own JWT. Kill it independently.
const search = clampd.guard(searchFn, { agentId: "research-agent", toolName: "web.search" });

Streaming Guard (opt-in)

 typescript
const client = clampd.openai(new OpenAI(), {
  agentId: "my-agent",
  guardStream: true,  // buffer + guard tool call chunks
});
// Text chunks stream instantly. Tool calls buffered → guarded → released.

MCP Proxy CLI

bash

npx tsx mcp-cli.ts \
  --downstream "npx -y @modelcontextprotocol/server-filesystem /tmp" \
  --agent-id "your-agent-id"

CLI Tool

clampd CLI

A standalone CLI for managing your Clampd cluster from the terminal. Inspect agents, trigger kills, import policies, stream audit logs, run compliance reports, and monitor everything with a live TUI dashboard.

Download now: Pre-built binaries for Linux (x86_64, ARM64), Windows, and static musl builds are available on the Setup page. macOS builds coming soon.

Commands

Command	Description
`clampd agent list`	List all registered agents with state and risk
`clampd kill <agent-id>`	Emergency kill switch
`clampd policy list`	Show active policies
`clampd watch`	Live TUI dashboard - real-time monitoring
`clampd demo`	Run live demo with real threat detection

Agent Management

 bash
# List all agents
$ clampd agent list
┌──────────┬──────────────┬────────┬──────┬───────────┐
│ ID       │ Name         │ State  │ Risk │ Framework │
├──────────┼──────────────┼────────┼──────┼───────────┤
│ a1b2c3.. │ demo-scout   │ active │ low  │ langchain │
│ d4e5f6.. │ demo-exec    │ active │ med  │ openai    │
│ g7h8i9.. │ attacker-bot │ killed │ high │ custom    │
└──────────┴──────────────┴────────┴──────┴───────────┘

# Kill a compromised agent
$ clampd kill g7h8i9.. --reason "Breach pattern detected"
Kill cascade initiated for agent g7h8i9..
  ✓ Agent terminated across all layers
  ✓ Agent state → killed

Live TUI Dashboard

 terminal
$ clampd watch
╔═ SERVICES ══════════╦═ AGENTS ══════════════════════╗
║ gateway       [OK]  ║ > demo-scout   active  low    ║
║ registry      [OK]  ║   demo-exec    active  med    ║
║ classifier    [OK]  ║   attacker-bot killed  high   ║
║ policy        [OK]  ║                              ║
║ token         [OK]  ║                              ║
╠═ LIVE EVENTS ═══════╩══════════════════════════════╣
║ 14:01:02 demo-scout   access_scoped  scope=read    ║
║ 14:01:03 demo-exec    tool_call api.stripe.charge ║
║ 14:01:04 demo-exec    POLICY_DENY over_budget      ║
║ 14:01:05 attacker-bot KILL_SWITCH breach_detected  ║
╠═ DETAILS ══════════════════════════════════════════╣
║ Selected: demo-scout | State: active               ║
║ Framework: langchain | Boundaries: max_calls=100   ║
╚════════════════════════════════════════════════════╝

Demo Mode

 bash
# Run breach scenario - registers real agents, sends real requests
# through the full pipeline, triggers real kill cascades
$ clampd demo --scenario breach

Starting demo scenario: breach
Checking cluster services...
Gateway reachable
Registering demo agents...
Agents ready. Launching TUI + demo events...

# The TUI shows live events as they flow through the pipeline:
# - Safe queries → ALLOW → forwarded to target
# - DROP TABLE / SSRF / reverse shell → BLOCKED by rules engine
# - Repeated violations → automatic kill cascade

Context Management

Manage multiple environments with named contexts - like kubectl config. Each context stores an endpoint, org ID, and credentials. Switch instantly between local, staging, and production.

 bash
# Initialize config
$ clampd config init
Config written to ~/.clampd/config.toml
Default context 'local' created.

# Add a production context
$ clampd context add prod \
    --endpoint https://api.clampd.dev \
    --gateway-url https://gateway.clampd.dev \
    --set-org-id a1234567-... \
    --set-api-token ag_live_xxx

# List contexts (* = active)
$ clampd context list
     NAME            DASHBOARD                                GATEWAY                        ORG_ID
 *   local           http://127.0.0.1:3001                    http://127.0.0.1:8080          (not set)
     prod            https://api.clampd.dev                   https://gateway.clampd.dev     a1234567-...
     staging         https://staging.internal:3001             https://staging.internal:8080   c9999999-...

# Switch context
$ clampd context use prod
Switched to context 'prod'.

# All commands now target prod
$ clampd agent list
$ clampd cluster status

Configuration File

Stored at ~/.clampd/config.toml. Contexts replace the old flat config - existing configs auto-migrate.

 ~/.clampd/config.toml
current_context = "local"

# Local development
[[contexts]]
name          = "local"
dashboard_url = "http://127.0.0.1:3001"
gateway_url   = "http://127.0.0.1:8080"
org_id        = ""
api_token     = ""
license_token = ""

# Production (SaaS or self-hosted)
[[contexts]]
name          = "prod"
dashboard_url = "https://api.clampd.dev"
gateway_url   = "https://gateway.clampd.dev"
org_id        = "a1234567-1234-1234-1234-123456789012"
api_token     = "ag_live_xxx"
license_token = "eyJ..."

# Environment variables override the active context
# CLAMPD_ORG_ID, CLAMPD_API_TOKEN, CLAMPD_DASHBOARD_URL, CLAMPD_GATEWAY_URL

Architecture

How Clampd Works

A multi-stage runtime security layer sits between your agents and their tools. Every tool call is authenticated, classified, policy-checked, and audited.

API Reference

Interactive API documentation is available at your Dashboard API:

api.clampd.dev/documentation

┌─────────────────┐     ┌──────────────────────────────────────────────────┐     ┌──────────────┐
│  LLM / Agent    │────▶│            Clampd Security Pipeline               │────▶│  Your Tool   │
│  (LangChain,    │     │                                                  │     │  (DB, API,   │
│   ADK, MCP)     │◀────│  Authenticate → Classify → Evaluate → Audit      │◀────│   File, MCP) │
└─────────────────┘     │  <25ms p95 latency                               │     └──────────────┘
                        └──────────────────────────────────────────────────┘

Pipeline Traces

Real Pipeline Output

Actual log output from Docker. See exactly what happens when a request flows through the proxy.

Blocked request - DROP TABLE users

 pipeline trace
[classify]   Rule matched, tool=database.query
[classify]   Intent classified: risk=high, classification=Malicious
[policy]     Policy evaluated: agent=b0000000..., action=Deny
[audit]      Event logged (1 event)

Allowed request - SELECT name FROM users

 pipeline trace
[classify]   Intent classified: risk=low, classification=Safe
[policy]     Policy evaluated: action=Allow
[scope]      Least-privilege access applied
[gateway]    Downstream responded: status=200, latency=10ms
[audit]      Event logged (1 event)

Rules Engine

What Gets Blocked

Comprehensive built-in rules detect common attack patterns against AI agents across 20 languages (English, Spanish, French, German, Chinese, Japanese, Korean, Arabic, Hindi, Russian, Turkish, Thai, Vietnamese, Portuguese, Italian, Dutch, Polish, Indonesian, Swedish, Czech). Region-specific PII detection for Aadhaar, PAN, NRIC, and more. Configurable risk thresholds. Shown below are example threat categories - the full rule set is extensive.

Category	Detects	Example	Risk
Destructive SQL	DROP, TRUNCATE, DELETE without WHERE	`DROP TABLE users`	Critical
Credential Access	Attempts to read secrets, keys, tokens	`.env, .ssh, credentials`	Critical
SSRF / Cloud Metadata	Internal network probing, metadata endpoints	`169.254.169.254/...`	High
Injection Attacks	SQL injection, NoSQL injection, LDAP, SSTI	`WHERE 1=1; DROP ...`	High
Prompt Injection	Attempts to override agent instructions	`ignore previous instructions`	High
PII Exfiltration	SSN, credit cards, personal data leaks	`SSN: *--****`	High
Shell Exploits	Reverse shells, dangerous commands, code exec	`rm -rf /, bash -i`	Critical
XSS / Path Traversal	Script injection, directory escape	`<script>, ../../etc`	High

+ many more categories including DNS exfiltration, crypto mining, agent impersonation, and more.

Audit Trail

Audit Log

Every event is recorded. Query your audit trail with standard SQL.

 sql
SELECT tool_name, blocked, assessed_risk, denial_reason
FROM audit_logs LIMIT 8

tool_name	blocked	assessed_risk	denial_reason
database.query	true	high	destructive SQL
database.query	false	low
file.read	true	high	path traversal
file.read	false	low
http.fetch	true	high	SSRF detected
http.fetch	false	low
shell.exec	true	critical	dangerous command
shell.exec	false	low

Deployment

Deployment Options

Clampd is available as a hosted platform today. Self-hosted deployment (Docker Compose + Helm) is coming soon.

Hosted Platform (Available Now)

Sign up at app.clampd.dev, register your agents, install the SDK, and start guarding tool calls. No infrastructure to manage.

Docker Compose (coming soon)

One command will spin up the entire stack - 10 hardened Rust microservices, dashboard, and all infrastructure. Shipping next week for licensed users.

Helm / Kubernetes (coming soon)

Production deployment on AKS, EKS, or GKE with TLS, cert-manager, external databases, and Ingress support.

Prometheus Metrics

ag-gateway exposes a /metrics endpoint in Prometheus exposition format. Scrape it with any Prometheus-compatible tool.

 metrics
agentguard_requests_total, agentguard_requests_denied_total, agentguard_requests_allowed_total,
agentguard_requests_flagged_total, agentguard_circuit_breaker_open_total, agentguard_cache_hits_total,
agentguard_latency_sum_us, agentguard_latency_count

Tool Integration

Validating Micro-Tokens

When Clampd approves a tool call, it mints a short-lived, single-use micro-token and forwards it to your tool service via the Authorization: Bearer <token> header. Your tool service should validate this token before executing any action.

Token Properties

Property	Description
`sub`	Agent ID that initiated the tool call
`scope`	Space-separated list of scopes the agent is allowed (e.g. `db:read`)
`exp`	Expiry timestamp - tokens expire in 30 seconds
`jti`	Unique token ID - each token can only be used once
`ag:tool_binding`	Hash binding the token to a specific tool descriptor
`ag:session_id`	Session ID (optional) for correlating calls within a session
`ag:trust_level`	Trust level (optional) - present when operating in degraded mode

Validation via Introspect API

The recommended approach is to call the Clampd Introspect endpoint, which verifies the signature, checks expiry, enforces single-use, and confirms the agent is not on the deny list - all in one call.

 pseudo-code
# In your tool service - before executing the tool action

def handle_tool_request(request):
    token = request.headers["Authorization"].removeprefix("Bearer ")

    # Call Clampd Introspect endpoint
    result = clampd_client.introspect(token)

    if not result.active:
        return error(403, "Token rejected: " + result.reason)

    # Check that the token grants the required scope
    if "db:read" not in result.scope.split():
        return error(403, "Insufficient scope")

    # Token is valid, single-use enforced, agent is active
    return execute_tool(request)

Introspect Response

Field	Type	Description
`active`	boolean	`true` if the token is valid and consumable
`sub`	string	Agent ID
`scope`	string	Space-separated scopes
`exp`	integer	Expiry (Unix timestamp)
`tool_binding`	string	Tool descriptor binding hash
`reason`	string?	Why the token was rejected (only when `active` is `false`)

Tokens are single-use: once introspected, the nonce is consumed and the same token cannot be used again. If your tool service does not call Introspect, the token expires automatically after 30 seconds.

Security Testing

External Security Testing

Validate your Clampd deployment with built-in testing tools, external red team frameworks, and community payload lists. Run attacks against your own cluster to verify every rule fires correctly.

Built-in Testing

The clampd test CLI command fires a suite of attack payloads against your gateway and reports which were blocked, flagged, or allowed.

Flag	Description
`--gateway <url>`	Gateway URL (default: `http://localhost:8080`)
`--attacks <category\|all>`	Attack categories to run: `sqli`, `xss`, `ssrf`, `shell`, `traversal`, `injection`, or `all`
`--json`	Output results as JSON (for CI pipelines)
`--watch`	Continuous mode — re-run tests on an interval
`--interval <seconds>`	Interval between watch runs (default: 60)
`--exit-on-fail`	Exit with non-zero code if any attack is not blocked (CI gate)

bash

# Run all built-in attacks, output JSON, fail if any attack gets through
clampd test --gateway http://localhost:8080 --attacks all --json --exit-on-fail

You can also use the Red Team Live tab in the Playground to run attacks interactively from the browser.

External Red Team Tools

Clampd endpoints are standard HTTP — any security testing tool that can send POST requests can target them. Below are well-known frameworks and payload sources that work out of the box.

Tool	Install	What It Tests	Clampd Endpoint
Promptfoo	`npm i -g promptfoo`	80+ attack plugins, encoding evasion	`/v1/proxy`, `/v1/scan-input`
Garak (NVIDIA)	`pip install garak`	Encoding probes, jailbreaks, DAN attacks	`/v1/scan-input`
PyRIT (Microsoft)	`pip install pyrit`	Multi-step attack chains, red teaming	`/v1/proxy`
SecLists	`github.com/danielmiessler/SecLists`	1000+ SQLi, XSS, LFI payloads	`/v1/scan-input`
PayloadBox	`github.com/payloadbox`	Categorized injection payloads	`/v1/scan-input`

Example: Promptfoo Config for Clampd

Save this as promptfooconfig.yaml and run promptfoo redteam run to fire 50 adversarial probes at your gateway.

 yaml
targets:
  - id: http
    config:
      url: 'http://localhost:8080/v1/scan-input'
      method: 'POST'
      headers:
        'Content-Type': 'application/json'
        'X-AG-Key': '${CLAMPD_API_KEY}'
        'Authorization': 'Bearer ${CLAMPD_JWT}'
      body:
        text: '{{prompt}}'
        agent_id: '${CLAMPD_AGENT_ID}'

redteam:
  purpose: 'AI agent tool call security'
  numTests: 50
  plugins:
    - 'prompt-injection'
    - 'sql-injection'
    - 'ssrf'
    - 'shell-injection'
  strategies:
    - 'jailbreak'
    - 'base64'
    - 'rot13'

Example: Garak Config for Clampd

Garak uses a REST generator to target any HTTP endpoint. Save as garak-clampd.yaml and run garak --model_type rest --model_name clampd --config garak-clampd.yaml.

 yaml
# garak-clampd.yaml - REST generator targeting Clampd scan-input
rest:
  uri: 'http://localhost:8080/v1/scan-input'
  method: POST
  headers:
    Content-Type: 'application/json'
    X-AG-Key: '${CLAMPD_API_KEY}'
    Authorization: 'Bearer ${CLAMPD_JWT}'
  body: '{"text": "$prompt", "agent_id": "${CLAMPD_AGENT_ID}"}'
  response_json_field: 'result'

# Run with encoding and jailbreak probes
# garak --model_type rest --model_name clampd --config garak-clampd.yaml \
#   --probes encoding,dan,gcg

Quick Payload Test Script

A minimal bash script to fire payload lists (SecLists, PayloadBox, or your own) at Clampd and report which payloads were not blocked.

 bash
#!/bin/bash
# test-clampd-payloads.sh - fire payload lists at Clampd
GW="${1:-http://localhost:8080}"
KEY="${CLAMPD_API_KEY:-ag_test_acme_2026}"
JWT="${CLAMPD_JWT}"
AGENT="${CLAMPD_AGENT_ID:-b0000001-0004-0000-0000-000000000001}"
FILE="${2:-payloads.txt}"

if [ ! -f "$FILE" ]; then
  echo "Usage: $0 [gateway-url] <payload-file>"
  echo "  e.g. $0 http://localhost:8080 SecLists/Fuzzing/SQLi/quick-SQLi.txt"
  exit 1
fi

passed=0; failed=0; total=0
while IFS= read -r payload; do
  [ -z "$payload" ] && continue
  total=$((total + 1))
  resp=$(curl -s -o /dev/null -w "%{http_code}" \
    -X POST "$GW/v1/scan-input" \
    -H "Content-Type: application/json" \
    -H "X-AG-Key: $KEY" \
    -H "Authorization: Bearer $JWT" \
    -d "{\"text\": $(echo "$payload" | jq -Rs .), \"agent_id\": \"$AGENT\"}")
  if [ "$resp" = "200" ]; then
    echo "MISS [$resp] $payload"
    failed=$((failed + 1))
  else
    passed=$((passed + 1))
  fi
done < "$FILE"

echo ""
echo "Results: $total tested, $passed blocked, $failed missed"
[ "$failed" -gt 0 ] && exit 1 || exit 0

Docs & SDK Reference

Python Quickstart

Guard OpenAI tool calls

Guard Anthropic / Claude

Guard any function

LangChain (callback - guards ALL tools)

Google ADK

Response Inspection (opt-in)

Multi-Agent (per-agent identity)

Streaming Guard (opt-in)

CrewAI

MCP Proxy (wrap any MCP server)

TypeScript Quickstart

Guard OpenAI tool calls

Guard Anthropic / Claude

Guard any function

Wrap OpenAI tool definitions

Response Inspection (opt-in)

Multi-Agent (per-agent identity)

Streaming Guard (opt-in)

MCP Proxy CLI

clampd CLI

Commands

Agent Management

Live TUI Dashboard

Demo Mode

Context Management

Configuration File

How Clampd Works

API Reference

Real Pipeline Output

Blocked request - DROP TABLE users

Allowed request - SELECT name FROM users

What Gets Blocked

Audit Log

Deployment Options

Hosted Platform (Available Now)

Docker Compose (coming soon)

Helm / Kubernetes (coming soon)

Prometheus Metrics

Validating Micro-Tokens

Token Properties

Validation via Introspect API

Introspect Response

External Security Testing

Built-in Testing

External Red Team Tools

Example: Promptfoo Config for Clampd

Example: Garak Config for Clampd

Quick Payload Test Script