Python Agents¶

import requests

RAMPART_URL = "http://localhost:9090"
RAMPART_TOKEN = "your-token"

def safe_exec(command: str) -> dict:
    """Check a command with Rampart before executing."""
    response = requests.post(
        f"{RAMPART_URL}/v1/tool/exec",
        headers={"Authorization": f"Bearer {RAMPART_TOKEN}"},
        json={
            "agent": "my-python-agent",
            "session": "session-1",
            "params": {"command": command}
        }
    )
    result = response.json()

    if result["decision"] == "deny":
        return {"blocked": True, "reason": result["message"]}

    # Command was allowed — execute it
    import subprocess
    output = subprocess.run(command, shell=True, capture_output=True, text=True)
    return {"blocked": False, "output": output.stdout}

def preflight(command: str) -> bool:
    """Check if a command is allowed without executing."""
    response = requests.post(
        f"{RAMPART_URL}/v1/preflight/exec",
        headers={"Authorization": f"Bearer {RAMPART_TOKEN}"},
        json={
            "agent": "my-agent",
            "session": "s1",
            "params": {"command": command}
        }
    )
    return response.json()["allowed"]

from langchain.tools import tool

@tool
def run_command(command: str) -> str:
    """Execute a shell command (Rampart-protected)."""
    resp = requests.post(
        f"{RAMPART_URL}/v1/tool/exec",
        headers={"Authorization": f"Bearer {RAMPART_TOKEN}"},
        json={"agent": "langchain", "session": "s1", "params": {"command": command}}
    )
    data = resp.json()
    if data["decision"] == "deny":
        return f"Command blocked by policy: {data['message']}"
    # Rampart only evaluates policy — execute the command yourself
    result = subprocess.run(command, shell=True, capture_output=True, text=True)
    return result.stdout