Python Agents¶

Integrate Rampart with any Python agent framework — LangChain, CrewAI, AutoGen, or custom code.

Python SDK (source-distributed alpha)¶

SDK 0.2.0 has its own component version, independent of the Rampart CLI. Its outage default intentionally differs from SDK 0.1.0; see the migration below.

Start the Rampart proxy:

rampart serve

In a second terminal, configure the SDK for that local service. The service saves its administrative token in ~/.rampart/token; the SDK reads RAMPART_TOKEN rather than automatically opening that file:

export RAMPART_URL="http://127.0.0.1:9090"
export RAMPART_TOKEN="$(cat "$HOME/.rampart/token")"

This loads the token without displaying it. Use these settings only for local development. For a deployed agent, provision an evaluation-scoped token and supply it through the agent's secret configuration. Remote endpoints require HTTPS and an explicitly configured token; keep token values out of code and logs.

The SDK is not currently published on PyPI. From a Rampart source checkout, install it and put its fail-closed enforcement call directly at the execution boundary:

python -m pip install ./sdks/python

import subprocess
import rampart

@rampart.exec_guard()
def safe_exec(command: str) -> str:
    return subprocess.check_output(command, shell=True, text=True)

The decorator sends an enforce: true preflight at the actual call boundary, consumes one-shot grants and call-count state exactly once, and does not execute the function unless Rampart returns a consistent allow decision. Its default client fails closed when the policy service is unavailable or malformed.

RampartClient() also fails closed by default, so supplying a custom client to change its URL or timeout preserves this protection. Transport failures and timeouts raise RampartConnectionError; HTTP server errors raise RampartServerError. Guards do not execute when these errors occur, including when raise_on_deny=False.

Upgrading earlier SDK source checkouts: calls that omitted fail_open previously returned a synthetic allow during transport/server failures. They now raise, including preflight, check_*, and their async equivalents. Handle availability exceptions without executing the guarded operation. If the application deliberately needs the previous availability fallback, use RampartClient(fail_open=True) explicitly. This does not override policy deny/ask or invalid decision responses. Explicit true/false settings and healthy decisions retain their behavior; health() and ahealth() still return False when unavailable.

Preflight API¶

Check if a command would be allowed without executing it:

import rampart

def preflight(command: str) -> bool:
    """Check if a command is allowed without executing."""
    with rampart.RampartClient(fail_open=False) as client:
        return client.check_exec(command, agent="my-agent", session="s1").allowed

Preflight is a preview only. Never separate a preflight from later execution; use enforce, aenforce, or a guard decorator at the execution boundary.

LD_PRELOAD Alternative¶

For simpler integration, wrap your entire Python process:

This optional path requires a source-built native library; current release and Homebrew packages contain only the Rampart CLI. Run make -C preload install from a matching Rampart source checkout first.

rampart preload -- python my_agent.py

This can intercept supported exec-family calls that reach the dynamic loader, including common os.system() and subprocess paths. It does not cover static binaries, direct syscalls, in-process Python file or network I/O, or every possible subprocess implementation.

LangChain Example¶

import subprocess
import rampart
from langchain.tools import tool

@tool
@rampart.exec_guard()
def run_command(command: str) -> str:
    """Execute a shell command only after Rampart authorizes this invocation."""
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout

API Reference¶