Metadata-Version: 2.4
Name: clampd
Version: 0.23.6
Summary: Runtime security SDK for AI agents — guard tool calls in 1 line
Project-URL: Homepage, https://clampd.dev
Project-URL: Repository, https://github.com/clampd/clampd
Author-email: Clampd <mehul.soni89@gmail.com>
License: Apache-2.0
Keywords: agents,ai,anthropic,crewai,function-calling,langchain,openai,proxy,security
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx<1.0,>=0.27
Requires-Dist: pydantic<3.0,>=2.0
Provides-Extra: all
Requires-Dist: anthropic<2.0,>=0.20; extra == 'all'
Requires-Dist: crewai<2.0,>=0.80; extra == 'all'
Requires-Dist: cryptography>=41.0; extra == 'all'
Requires-Dist: langchain-core<1.0,>=0.3; extra == 'all'
Requires-Dist: mcp<2.0,>=1.0; extra == 'all'
Requires-Dist: openai<3.0,>=1.0; extra == 'all'
Provides-Extra: anthropic
Requires-Dist: anthropic<2.0,>=0.20; extra == 'anthropic'
Provides-Extra: crewai
Requires-Dist: crewai<2.0,>=0.80; extra == 'crewai'
Provides-Extra: dev
Requires-Dist: cryptography>=41.0; extra == 'dev'
Requires-Dist: mypy>=1.0.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.21.0; extra == 'dev'
Requires-Dist: pytest-cov>=4.0.0; extra == 'dev'
Requires-Dist: pytest>=7.0.0; extra == 'dev'
Requires-Dist: ruff>=0.1.0; extra == 'dev'
Provides-Extra: langchain
Requires-Dist: langchain-core<1.0,>=0.3; extra == 'langchain'
Provides-Extra: mcp
Requires-Dist: mcp<2.0,>=1.0; extra == 'mcp'
Provides-Extra: openai
Requires-Dist: openai<3.0,>=1.0; extra == 'openai'
Provides-Extra: verify
Requires-Dist: cryptography>=41.0; extra == 'verify'
Description-Content-Type: text/markdown

# 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

```bash
pip install clampd
```

With framework extras:

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

## Quick Start

```python
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 hands the LLM a structured hint
the model can pattern-match on instead of a free-text "denied" string.
0.23.3 is the cleanup: every corrective comes from a rule or policy,
no code-side overrides, no synthesized fallbacks.

- **Rule-only correctives.** The SDK no longer ships a `suggest=`
  kwarg. Every corrective comes from a rule's `[rule.corrective]`
  block or a Cedar `@corrective_*` annotation, which means security
  policy lives in the admin surface, not in tool-author code.

- **Honest fallbacks.** When no source authors a corrective, denials
  emit `kind = "no_correction"` with rule attribution rather than a
  synthesized scope-mismatch hint that may be semantically wrong (the
  old "tool requires X / Permitted: X / Closest: X" bug is gone).

- **`request_approval` variant removed** from the public surface
  until admin approval routing is built. The proto wire shape keeps
  the variant for backwards compat; SDKs simply don't emit or parse
  it any more.

- **Typed corrective actions.** Denials carry one of **9 variant
  shapes** (`switch_tool`, `downscope_to`, `rename_field`,
  `redact_value`, `split_request`, `wait_and_retry`, `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.

- **`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 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 and ~180 other detection rules now emit
  `kind = "no_correction"` with an empty hint, so the LLM-facing
  string is the bare phrase `"Action blocked."` — nothing for an
  attacker to iterate on. The dashboard chip still renders for
  operator visibility.

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

```python
# 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

```python
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

```python
import clampd

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

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

## Google ADK

```python
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)

```python
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)

```python
# 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

```python
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)

```python
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.

```python
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

```python
# 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.

```python
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](https://clampd.dev) gateway

## License

BUSL-1.1
