clampd 0.23.3

Runtime security SDK for AI agents — guard tool calls in 1 line

Clampd Python SDK

Runtime security for AI agents. Guard every tool call — OpenAI, Anthropic, LangChain, Google ADK — in 1 line. Prompt and response scanning enabled by default.

Installation

pip install clampd

With framework extras:

pip install clampd[langchain]    # LangChain callback handler
pip install clampd[mcp]          # MCP server support
pip install clampd[all]          # Everything

Quick Start

import clampd
from openai import OpenAI

# Configure once at startup
clampd.init(
    agent_id="my-agent",
    secret="ags_...",              # from dashboard → Agent → Secret
    gateway_url="http://localhost:8080",
    api_key="ag_live_...",
)

# Wrap your OpenAI client — done
client = clampd.openai(OpenAI())

# Use it exactly like before. Clampd intercepts every tool call.
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Look up active users"}],
    tools=[...],
)
# Dangerous tool calls → blocked before execution
# Safe tool calls → proceed normally
# Prompts scanned before LLM, responses scanned after

What's New in 0.23.3

When Clampd blocks a tool call, it now hands the LLM a structured hint instead of a free-text "denied" string. The model can pattern-match on the hint and pivot, often without going back to the user.

A few highlights:

• Rule-only correctives. Removed the SDK-side suggest= kwarg and the request_approval variant. Every corrective now comes from a rule's [rule.corrective] block or a Cedar @corrective_* annotation.

• Honest fallbacks. When no rule authored a corrective, denials emit kind = "no_correction" with rule attribution rather than a synthesized scope-mismatch hint that may be semantically wrong.

• Typed corrective actions. Denials carry one of 10 variant shapes (switch_tool, downscope_to, rename_field, redact_value, split_request, wait_and_retry, request_approval, switch_endpoint, no_correction, plus downscope_auto for resolver-picked alternatives). Read error.denial.corrective for the typed shape; call error.to_tool_result() for the ready-to-send string.

• @clampd.guard(suggest=...). If you know your tool best, pin the remedy on the decorator. Your hint wins over the rule defaults and Cedar templates.

• ClampdLoopError. When an LLM keeps retrying the same denied call (idempotency key seen three times in a row), this is raised instead of another ClampdBlockedError. Catch it first so loop detection isn't swallowed.

• clampd.register_tool(). Declare each tool's category at startup. Bypasses default-deny on first use and locks the descriptor hash so rug-pull detection has a baseline.

• Bard-quality messages. Every denial now reads Action blocked: X. Reformulate the call under scope \Y``. The next step lives in backticks where the LLM can grab it cleanly.

• Silent on attacks. Prompt-injection, command-injection, RCE, SSRF, path-traversal — Clampd no longer hands the model a "try scope Y instead" hint when those signals fire. The dashboard still shows the chip; the LLM-facing string is blank so an attacker can't iterate.

What's New in 0.5.0

• Per-agent JWT identity — each agent authenticates independently in multi-agent systems
• Streaming guard — opt-in tool call interception for streaming responses (guard_stream=True)
• Circuit breaker & retry — automatic retry with exponential backoff
• CrewAI integration — guard CrewAI agent tool calls
• 216 detection rules with Aho-Corasick prefilter (22μs at 10K rules)

Configuration

# Option 1: Single agent (simple)
clampd.init(
    agent_id="my-agent",
    secret="ags_...",
    gateway_url="http://localhost:8080",
    api_key="ag_live_...",
)

# Option 2: Multi-agent (per-agent identity)
clampd.init(
    agent_id="orchestrator",
    api_key="ag_live_...",
    agents={
        "orchestrator": os.environ["CLAMPD_SECRET_orchestrator"],
        "research-agent": os.environ["CLAMPD_SECRET_research_agent"],
        "writer-agent": os.environ["CLAMPD_SECRET_writer_agent"],
    },
)

# Option 3: Environment variables
# CLAMPD_GATEWAY_URL=http://localhost:8080
# CLAMPD_API_KEY=ag_live_...
# CLAMPD_SECRET_orchestrator=ags_...
# CLAMPD_SECRET_research_agent=ags_...

Anthropic / Claude

import clampd
from anthropic import Anthropic

clampd.init(agent_id="my-agent", secret="ags_...")
client = clampd.anthropic(Anthropic())

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "..."}],
    tools=[...],
)

LangChain

import clampd

handler = clampd.langchain(agent_id="my-agent", secret="ags_...")

result = executor.invoke(
    {"input": "Look up active users"},
    config={"callbacks": [handler]},
)

Google ADK

import clampd
from google.adk import Agent

agent = Agent(
    tools=[...],
    before_tool_callback=clampd.adk(agent_id="my-agent", secret="ags_..."),
)

Multi-Agent (A2A Delegation)

import os
import clampd

# Each agent authenticates with its own secret.
# Delegation chains are tracked automatically.
clampd.init(
    agent_id="orchestrator",
    api_key="ag_live_...",
    agents={
        "orchestrator": os.environ["CLAMPD_SECRET_orchestrator"],
        "research-agent": os.environ["CLAMPD_SECRET_research_agent"],
    },
)

# research-agent gets its own JWT (sub=research-agent).
# Kill "research-agent" from dashboard → only this agent is blocked.
@clampd.guard("web.search", agent_id="research-agent")
def search(query: str):
    return web_search(query)

Streaming Guard (opt-in)

# Stream tool calls are guarded only when guard_stream is enabled.
client = clampd.openai(OpenAI(),
    agent_id="my-agent",
    guard_stream=True,  # buffer + guard tool call chunks before release
)

stream = client.chat.completions.create(
    model="gpt-4o",
    stream=True,
    tools=[...],
    messages=[{"role": "user", "content": "..."}],
)
# Tool calls in the stream are buffered, guarded, then released.
# Text chunks pass through immediately with zero added latency.

CrewAI

import clampd
from clampd.crewai_callback import ClampdCrewAIGuard

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

# Wrap CrewAI tools
safe_tool = guard.wrap_tool(my_tool)

Direct Guard (any function)

import clampd

clampd.init(agent_id="my-agent", secret="ags_...")

@clampd.guard("database.query")
def run_query(sql: str):
    return db.execute(sql)

# With response checking (opt-in)
@clampd.guard("file_read", check_response=True)
def read_file(path: str):
    return open(path).read()

run_query("SELECT * FROM users")     # allowed
run_query("DROP TABLE users")        # raises ClampdBlockedError

Tool Registration (recommended at startup)

Declare each tool's category once. Tools registered this way bypass default-deny on first use and lock their descriptor hash so rug-pull detection has a baseline to compare against.

import clampd

clampd.init(agent_id="my-agent", secret="ags_...")

# Declare tool classification once at startup
clampd.register_tool(
    "database.query",
    category=clampd.Category.DB,
    subcategory=clampd.Subcategory.QUERY,
    operation=clampd.Operation.READ,
    description="Read-only SQL against the analytics DB.",
)

@clampd.guard("database.query")
def database_query(sql: str): ...

You can also pass a framework tool object directly — LangChain BaseTool, OpenAI tool dict, or Anthropic tool dict — and Clampd extracts the name and schema.

How corrective hints get to the LLM

You don't have to do anything. When a tool call gets denied, Clampd returns a typed hint (switch_tool → archive_table, wait_and_retry, etc.) that the LLM can pattern-match on. The hint comes from whichever source matched first:

boundary > sdk_override > cedar > per-agent > rule template > downscope_auto

For most rules this is already wired. R001 (destructive SQL) for example ships with a switch_tool corrective pointing at archive_table. When the LLM hits that, it sees:

Action blocked: Destructive SQL (DROP/TRUNCATE/DELETE) is blocked.
Use archive_table to soft-delete (archived=true column).
Reformulate this call using the `archive_table` tool instead.

It pattern-matches on `archive_table` and pivots on the next turn.

Authoring custom correctives

The recommended path is the dashboard. Two ways:

1. Cedar policy with @corrective_* annotations. Author once, covers every agent in the org.
2. Per-agent override on the rules page. Useful when one agent needs a different remedy than the org default.

Both ship in 0.23.0+ via the Policies / Agents UI.

Valid kind values: switch_tool, downscope_to, downscope_auto, rename_field, redact_value, split_request, wait_and_retry, switch_endpoint, no_correction.

Authoring is now the only path. As of v0.23.3 Clampd no longer accepts code-side corrective overrides via the SDK. Correctives must be authored on the rule (via TOML or the dashboard) or on the Cedar policy. This puts security policy where it belongs — under admin review — and removes a class of override that bypassed the audit trail.

Scanning Options

# Defaults (v0.4.0+): scan_input=True, scan_output=True
client = clampd.openai(OpenAI(), agent_id="my-agent")

# Opt out of scanning
client = clampd.openai(OpenAI(),
    agent_id="my-agent",
    scan_input=False,   # skip prompt scanning
    scan_output=False,  # skip response scanning
)

Error Handling

As of v0.20+, blocked tool calls carry a typed StructuredDenial with a corrective action the LLM can pattern-match on. Catch ClampdLoopError before ClampdBlockedError so legitimate loop detection isn't swallowed by the more general handler.

from clampd import ClampdBlockedError, ClampdLoopError

try:
    run_query("DROP TABLE users")
except ClampdLoopError as e:
    # The LLM has retried the same denied call too many times.
    # Surface as a hard error — don't feed back to the model.
    raise
except ClampdBlockedError as e:
    # Hand the gateway-rendered string back to the LLM tool loop —
    # the model will pattern-match on the backticked tool / scope.
    tool_result_content = e.to_tool_result()
    # Or inspect the typed corrective directly:
    if e.denial and e.denial.corrective:
        c = e.denial.corrective
        match c.action:
            case clampd._corrective.SwitchTool(tool=t):
                # Auto-retry with the safer tool
                ...
            case clampd._corrective.WaitAndRetry(retry_after_seconds=n):
                # Sleep + retry
                ...

error.denial is StructuredDenial | None carrying:

• rule_id — the rule or NEVER_EXEMPTABLE predicate that fired
• violated_predicate — human-readable WHY (e.g. "Destructive SQL blocked")
• corrective — typed CorrectiveAction or None
• idempotency_key — stable hash so the SDK can detect loops
• reason_codes, boundary_violation

error.to_tool_result() returns the gateway's pre-rendered string ready to drop into tool_result.content — same text the dashboard chip shows, no client-side template logic.

API Reference

Function	Description
clampd.init(...)	Configure global client. agents= for per-agent secrets.
clampd.register_tool(name, category, subcategory, operation, ...)	Declare a tool's taxonomy classification at startup. Bypasses default-deny on first use.
clampd.openai(client, **opts)	Wrap OpenAI client. guard_stream=True for streaming.
clampd.anthropic(client, **opts)	Wrap Anthropic client. guard_stream=True for streaming.
clampd.guard(tool_name, **opts)	Decorator for any function. Correctives are authored on the rule (TOML or dashboard) or Cedar policy.
clampd.langchain(...)	LangChain callback handler.
clampd.adk(...)	Google ADK before_tool_callback.
ClampdCrewAIGuard	CrewAI tool wrapping.
clampd.delegation_headers() / enter_delegation(...)	A2A delegation propagation.
clampd.verify_scope_token(...) / require_scope(...)	Tool-side scope-token verification (zero-trust).
clampd.Category / Subcategory / Operation	Taxonomy enums for register_tool.
ClampdBlockedError / ClampdLoopError / ClampdUnregisteredToolError	Typed exception hierarchy.

Requirements

• Python 3.10+
• A running Clampd gateway

License

BUSL-1.1