cordum-guard

Safety governance for Python AI agents. Add Cordum safety checks to existing Python code with a single decorator — no rewrite required.

Install

pip install cordum-guard

With LangChain or LlamaIndex support:

pip install cordum-guard[langchain]
pip install cordum-guard[llamaindex]

Quick Start

from cordum_guard import CordumClient, guard

client = CordumClient("http://localhost:8081", api_key="your-api-key")

@guard(client, policy="financial_ops", risk_tags=["write", "financial"])
def execute_transfer(amount: float, to_account: str):
    bank_api.transfer(amount, to_account)

The @guard decorator intercepts every call to execute_transfer:

1. Evaluates the safety policy via the Cordum Safety Kernel
2. allow — function runs normally
3. deny — raises CordumBlockedError
4. require_approval — waits for human approval in the dashboard
5. throttle — delays execution per policy

LangChain Integration

from cordum_guard import CordumClient
from cordum_guard.langchain import CordumToolGuard

client = CordumClient("http://localhost:8081", api_key="your-api-key")
guarded_tools = CordumToolGuard(client, policy="agent_ops").wrap(tools)
agent = initialize_agent(guarded_tools, llm)

LlamaIndex Integration

from cordum_guard import CordumClient
from cordum_guard.llamaindex import CordumToolGuard

client = CordumClient("http://localhost:8081", api_key="your-api-key")
guarded_tools = CordumToolGuard(client, policy="agent_ops").wrap(tools)

Async Support

The @guard decorator works with both sync and async functions:

@guard(client, policy="ops")
async def async_operation():
    return await some_api_call()

Configuration

CordumClient

@guard decorator

CordumToolGuard (LangChain / LlamaIndex)

Testing

Use MockCordumClient to test guarded code without a live gateway:

from cordum_guard import guard, MockCordumClient, Decision

mock = MockCordumClient(default_decision=Decision.ALLOW)
mock.set_policy_response("dangerous-ops", Decision.DENY)

@guard(mock, capability="safe-op")
def safe_func():
    return "works"

@guard(mock, capability="dangerous-ops")
def risky_func():
    return "blocked"

assert safe_func() == "works"
# risky_func() raises CordumBlockedError

# Inspect what was evaluated:
assert len(mock.call_log) == 1
assert mock.call_log[0].capability == "safe-op"

Caching

Enable TTL-based caching to reduce gateway round-trips:

client = CordumClient(
    gateway_url="http://localhost:8081",
    api_key="my-key",
    cache_ttl=30,       # cache decisions for 30s (0 = disabled)
    cache_max_size=500,  # max cached entries
)
# ALLOW/DENY/THROTTLE cached; REQUIRE_APPROVAL always fresh
# Bypass cache per-call: client.evaluate_policy(..., cache=False)
# Clear cache: client.clear_cache()

Failure Modes

Configure what happens when the gateway is unreachable:

# Default: fail-closed (raise CordumConnectionError)
client = CordumClient("http://localhost:8081", api_key="key", on_error="closed")

# Fail-open: allow operations when gateway is down
client = CordumClient("http://localhost:8081", api_key="key", on_error="open")

# Callback: custom logic per error
def my_fallback(error):
    if "critical" in str(error):
        raise error  # fail closed for critical
    return SafetyDecision(decision=Decision.ALLOW)

client = CordumClient("http://localhost:8081", api_key="key", on_error=my_fallback)

Only connection/timeout errors trigger fail-open. Auth errors (401/403) and explicit DENY responses always propagate normally.

License

Apache-2.0