Python SDK
Arden documentation
Policy enforcement and human approval for AI agent tool calls. Install once, intercept everything.
Overview
Introduction
Arden sits between your AI agent and its tools. Every tool call is evaluated against policies you configure in the dashboard before the function runs. The policy engine returns one of three decisions:
Function executes immediately and returns its result.
PolicyDeniedError raised. Function never runs.
Paused until a human approves or denies on the dashboard.
Every call is logged regardless of outcome — you get a full audit trail from day one, even for tools with no policy configured.
Getting started
Quick start
Install the SDK:
pip install ardenpyGet your API key from app.arden.sh. Keys starting with arden_test_ hit the test environment; arden_live_ keys hit production.
Configure once at startup:
import ardenpy as arden
arden.configure(api_key="arden_live_...")
# For LangChain, CrewAI, and OpenAI Agents SDK — that's it.
# Every tool call in this process is now intercepted, enforced, and logged.For custom agents without a framework, wrap functions explicitly with guard_tool():
safe_refund = arden.guard_tool("issue_refund", issue_refund)
result = safe_refund(150.0, customer_id="cus_abc")
# → allowed · blocked · or held for human approvalConcepts
How it works
Every tool call follows the same sequence:
- 1The agent calls a tool (e.g.
issue_refund). - 2Arden intercepts the call and sends the tool name and all arguments to the policy engine.
- 3The engine looks up the policy for this agent and tool. If no policy exists, the call is allowed and logged.
- 4For requires_approval, the call pauses. A human approves or denies on the dashboard. For allow, the function runs immediately. For block,
PolicyDeniedErroris raised.
Tip
Integrations
Framework integrations
The integration path depends on which framework your agent uses.
LangChain and CrewAI — zero wrapping required
Arden automatically patches BaseTool.run at the class level when configure() is called. Every tool call in the process is intercepted — including tools created after configure is called.
pip install ardenpyimport ardenpy as arden
arden.configure(api_key="arden_live_...")
# Done. No protect_tools(), no guard_tool(), no imports.
# LangChain — use tools exactly as before
from langchain.tools import Tool
tools = [Tool(name="issue_refund", func=issue_refund, description="...")]
agent = create_react_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
executor.invoke({"input": user_message})
# Every tool the agent calls is policy-checked automatically# CrewAI — define BaseTool subclasses as usual
from crewai.tools import BaseTool
class RefundTool(BaseTool):
name: str = "issue_refund"
description: str = "Issue a refund to a customer"
def _run(self, amount: float, customer_id: str) -> str:
return f"Refund of {amount} issued"
agent = Agent(role="Support", tools=[RefundTool()], ...)
# Every _run() call is intercepted — no changes to tool classes neededTool names in the dashboard match each tool's .name attribute directly (e.g. issue_refund). Your API key already identifies which agent is making the call, so no prefix is needed.
Note
guard_tool() are automatically skipped by the auto-patcher — no double policy checks.OpenAI
The OpenAI Agents SDK is auto-patched by configure() — no extra wrapping needed. Arden patches FunctionTool.__init__ so every tool created with @function_tool is intercepted automatically.
pip install ardenpyimport ardenpy as arden
from agents import Agent, function_tool, Runner
arden.configure(api_key="arden_live_...")
@function_tool
def issue_refund(amount: float, order_id: str) -> str:
return f"Refund of {amount} issued for {"{order_id}"}"
# No wrapping needed — configure() already intercepted this tool
agent = Agent(name="SupportBot", tools=[issue_refund])
result = await Runner.run(agent, "Refund order FF-4210 for $500")Note
issue_refund).For the OpenAI Chat Completions API (raw tool-call loop), use ArdenToolExecutor:
# OpenAI Chat Completions loop
from ardenpy.integrations.openai import ArdenToolExecutor
executor = ArdenToolExecutor()
executor.register("issue_refund", issue_refund)
executor.register("send_email", send_email)
# In your tool-call loop:
result = executor.run(tc.function.name, json.loads(tc.function.arguments))Custom agents
For agents built without a framework, wrap each function with guard_tool(). The first argument is the tool name used for policy lookup in the dashboard.
import ardenpy as arden
arden.configure(api_key="arden_live_...")
safe_refund = arden.guard_tool("issue_refund", issue_refund)
safe_email = arden.guard_tool("send_email", send_email)
result = safe_refund(150.0, customer_id="cus_abc")API reference
configure()
configure()
ardenpy.configure(
api_key: str | None = None, # or ARDEN_API_KEY env var
environment: str | None = None, # "live" | "test" — auto-detected from key
timeout: float | None = None, # default 30.0 s
poll_interval: float | None = None, # default 2.0 s
max_poll_time: float | None = None, # default 300.0 s
retry_attempts: int | None = None, # default 3
signing_key: str | None = None, # for webhook signature verification
) -> ArdenConfigSets the global SDK configuration and auto-patches installed frameworks (LangChain, CrewAI, OpenAI Agents SDK). Call this once at application startup — every tool call in the process is intercepted, enforced, and logged from that point on.
| Parameter | Type | Default | Description |
|---|---|---|---|
| api_key | str | ARDEN_API_KEY | Your agent's API key. Keys prefixed arden_test_ automatically select the test environment. |
| environment | str | "live" | Override the environment. Auto-detected from the API key prefix when possible. |
| timeout | float | 30.0 | Seconds before an HTTP request to the Arden API times out. |
| poll_interval | float | 2.0 | Seconds between status polls when waiting for human approval. |
| max_poll_time | float | 300.0 | Maximum seconds to wait for a decision before raising ApprovalTimeoutError. |
| retry_attempts | int | 3 | Times to retry a failed request before raising ArdenError. |
| signing_key | str | None | Webhook signing key, used by verify_webhook_signature() to authenticate incoming webhook payloads. |
Warning
configure() a second time replaces the previous configuration globally.guard_tool()
ardenpy.guard_tool(
tool_name: str,
func: Callable,
approval_mode: str = "wait",
on_approval: Callable | None = None,
on_denial: Callable | None = None,
) -> CallableWraps func with Arden policy enforcement. Returns a new callable with the same signature — the original function is not modified. Use this for custom agents; for LangChain and CrewAI, configure() alone is sufficient.
| Parameter | Type | Default | Description |
|---|---|---|---|
| tool_name | str | — | Name used to look up this tool's policy in the dashboard. Set this to match the policy you create. |
| func | Callable | — | The function to protect. |
| approval_mode | str | "wait" | How to handle a requires_approval decision. One of "wait", "async", or "webhook". |
| on_approval | Callable | None | Required for "async" and "webhook" modes. Called with a WebhookEvent when approved. |
| on_denial | Callable | None | Required for "async" and "webhook" modes. Called with a WebhookEvent when denied. |
Arguments are captured using inspect.signature and bound_args.arguments, so positional calls like safe_refund(50.0) correctly send {"amount": 50.0} to the policy engine for rule evaluation.
handle_webhook()
ardenpy.handle_webhook(
body: bytes,
headers: dict[str, str],
signing_key: str | None = None,
) -> NoneVerifies a webhook payload, parses it into a WebhookEvent, and dispatches it to any callbacks registered via guard_tool(on_approval=..., on_denial=...). Raises ArdenError if the signature is invalid or the payload cannot be parsed.
| Parameter | Type | Default | Description |
|---|---|---|---|
| body | bytes | — | Raw request body bytes. |
| headers | dict[str, str] | — | Dict of request headers. Must include X-Arden-Signature and X-Arden-Timestamp. |
| signing_key | str | None | None | Overrides the signing key set in configure(). If omitted, uses configure(signing_key=...). |
verify_webhook_signature()
ardenpy.verify_webhook_signature(
body: bytes,
timestamp: str,
signature: str,
signing_key: str,
) -> boolLower-level signature verification. Returns True if valid. Use this if you want to verify the signature manually without calling handle_webhook().
| Parameter | Type | Default | Description |
|---|---|---|---|
| body | bytes | — | Raw request body bytes. |
| timestamp | str | — | X-Arden-Timestamp header value. |
| signature | str | — | X-Arden-Signature header value (e.g. sha256=abc123). |
| signing_key | str | — | Your webhook signing key from the dashboard. |
set_session()
ardenpy.set_session(session_id: str) -> None ardenpy.get_session() -> str | None ardenpy.clear_session() -> None
Attaches a session ID to the current async task or thread. Every policy-check call made after this point will include the session ID in the action record, allowing you to group and replay all actions from a single conversation on the dashboard.
| Parameter | Type | Default | Description |
|---|---|---|---|
| session_id | str | — | Any string identifying this session — a UUID, conversation ID, user ID, etc. Use get_session() to read it back; clear_session() to unset. |
Note
contextvars.ContextVar — safe for concurrent async requests with no locking required. See the Session tracking section for a full usage example.Concepts
Approval modes
When a policy returns requires_approval, the SDK's behavior is controlled by the approval_mode you set on guard_tool() (or on protect_tools() if using explicit wrapping).
wait (default)
safe_refund = arden.guard_tool("issue_refund", issue_refund)
# approval_mode="wait" is the defaultBlocks the calling thread and polls the Arden API every poll_interval seconds. Execution resumes once a human acts on the dashboard.
Approved
Function executes, result returned.
Denied
PolicyDeniedError raised.
Timeout
ApprovalTimeoutError raised.
Tip
async
safe_refund = arden.guard_tool(
"issue_refund", issue_refund,
approval_mode="async",
on_approval=handle_approved,
on_denial=handle_denied,
)
result = safe_refund(150.0, customer_id="cus_abc")
# Returns PendingApproval immediately — never blocks
# on_approval / on_denial called from a background thread when a human actsTip
webhook
safe_refund = arden.guard_tool(
"issue_refund", issue_refund,
approval_mode="webhook",
on_approval=handle_approved,
on_denial=handle_denied,
)
result = safe_refund(150.0, customer_id="cus_abc")
# Returns PendingApproval immediately
# Arden backend POSTs to your webhook URL when a human decidesTip
Observability
Audit trail
Every tool call — regardless of policy outcome — is written to your agent's action log. This happens automatically once you call configure(). No additional setup required.
| Decision | Logged? | Notes |
|---|---|---|
| allow | Yes | Includes all arguments and the matching policy. |
| allow (no policy) | Yes | reason: "no_policy_configured" — gives visibility before you add policies. |
| block | Yes | Includes arguments. Function never ran. |
| requires_approval | Yes | Action record created; updated when a human decides. |
View the action log for any agent at app.arden.sh. You can also check a specific action's status via the API:
curl 'https://api.arden.sh/status/<action_id>' \
-H 'X-API-Key: arden_live_...'Observability
Token usage tracking
Arden automatically tracks token consumption and estimated cost for every LLM call. For LangChain, CrewAI, and the OpenAI Agents SDK, configure() patches the framework at startup — no extra setup needed. Usage is sent in a background thread and adds no latency to your agent.
| Framework | Automatic? | How |
|---|---|---|
| LangChain | Yes | Patches BaseChatModel.invoke |
| CrewAI | Yes | Uses LangChain under the hood |
| OpenAI Agents SDK | Yes | Patches Runner.run |
| OpenAI Chat Completions | Manual | Call arden.log_token_usage() |
| Any other LLM | Manual | Call arden.log_token_usage() |
For custom agent loops, call log_token_usage() manually after each LLM response:
import ardenpy as arden
arden.configure(api_key="arden_live_...")
response = openai_client.chat.completions.create(
model="gpt-4o",
messages=messages,
)
arden.log_token_usage(
model="gpt-4o",
prompt_tokens=response.usage.prompt_tokens,
completion_tokens=response.usage.completion_tokens,
)ardenpy.log_token_usage(model, prompt_tokens, completion_tokens, session_id=None)
| Parameter | Type | Default | Description |
|---|---|---|---|
| model | str | — | Model name, e.g. "gpt-4o" or "claude-sonnet-4-6". Unknown models are logged with zero estimated cost. |
| prompt_tokens | int | — | Number of input tokens consumed. |
| completion_tokens | int | — | Number of output tokens generated. |
| session_id | str | None | — | Optional. Falls back to the current set_session() value if not provided. |
Note
The dashboard shows cost broken down by model, day, and session — all scoped per agent. View it at app.arden.sh under the Usage tab for each agent.
Observability
Session tracking
Attach a session ID to a conversation or workflow run so every tool call from that session is grouped together in the action log. Use this to replay exactly what an agent did in a specific conversation — useful for debugging, auditing, and customer support.
Call arden.set_session() once at the start of each request or conversation turn. All subsequent policy-check calls in that async task or thread automatically carry the session ID.
import ardenpy as arden
import uuid
arden.configure(api_key="arden_live_...")
# FastAPI example — set once per request
@app.post("/chat")
async def chat(request: Request):
body = await request.json()
# Use an ID your app already tracks, or generate one
session_id = body.get("session_id") or str(uuid.uuid4())
arden.set_session(session_id)
reply = await run_agent(messages)
return JSONResponse({"reply": reply, "session_id": session_id})The session ID is stored in a ContextVar, so it is scoped to the current async task. Concurrent requests never share a session ID, even without any extra locking.
Tip
set_session() is never called, nothing changes — existing integrations are unaffected.Session ID shows up as a top-level field on every action record in the log:
| Field | Description |
|---|---|
| session_id | The value passed to set_session(). Omitted if never called. |
| action_id | Unique ID for this specific tool call. |
| tool_name | Name of the tool that was called. |
| decision | allow · block · requires_approval |
| context | All arguments captured at call time. |
ardenpy.set_session(session_id: str) -> None ardenpy.get_session() -> str | None ardenpy.clear_session() -> None
| Parameter | Type | Default | Description |
|---|---|---|---|
| session_id | str | — | Any string identifying this session — a UUID, conversation ID, user ID, etc. |
Webhooks
Webhook setup
When using approval_mode="webhook", Arden POSTs a signed payload to your endpoint when a human approves or denies. Configure your webhook URL in the dashboard under agent settings.
Payloads are signed with HMAC-SHA256. Verify the signature before processing:
# FastAPI
from fastapi import FastAPI, Request
import ardenpy as arden
app = FastAPI()
@app.post("/arden/webhook")
async def webhook(request: Request):
body = await request.body()
arden.handle_webhook(
body=body,
headers=dict(request.headers),
)
# event_type is "action_approved" or "action_denied"
# callbacks registered via guard_tool(on_approval=...) are called automatically
return {"ok": True}# Flask
from flask import Flask, request
import ardenpy as arden
app = Flask(__name__)
@app.post("/arden/webhook")
def webhook():
arden.handle_webhook(
body=request.get_data(),
headers=dict(request.headers),
)
return {"ok": True}Note
signing_key in configure() to enable signature verification. Get your signing key from the dashboard under webhook settings.Error handling
Error handling
import ardenpy as arden
try:
result = safe_refund(150.0, customer_id="cus_abc")
except arden.PolicyDeniedError:
# Blocked by policy, or denied by a human (wait mode)
return {"ok": False, "reason": "not_allowed"}
except arden.ApprovalTimeoutError as e:
# Nobody approved within max_poll_time
return {"ok": False, "reason": "timeout", "action_id": e.action_id}
except arden.ArdenError as e:
# API error, configuration error, etc.
logger.error("Arden error: %s", e)
return {"ok": False, "reason": "policy_check_failed"}For async and webhook modes, only PolicyDeniedError (immediate block) and ArdenError can be raised at call time. ApprovalTimeoutError only applies to wait mode.
Tip
ArdenError broadly and failing open (allowing the call) or failing closed (blocking) depending on your risk tolerance.Configuration
Environments
Arden has two isolated environments — test and live. Each has its own API keys, agents, policies, and action log.
| Environment | API endpoint | Key prefix |
|---|---|---|
| Test | https://api-test.arden.sh | arden_test_ |
| Live | https://api.arden.sh | arden_live_ |
The environment is auto-detected from the API key prefix. Use an environment variable to switch without code changes:
import os, ardenpy as arden
arden.configure(api_key=os.environ["ARDEN_API_KEY"])
# ARDEN_API_KEY=arden_test_... → test environment
# ARDEN_API_KEY=arden_live_... → live environmentReference
Exceptions
All exceptions inherit from ArdenError.
class ArdenError(Exception)
Base class. Catch this to handle any Arden-related error.
class PolicyDeniedError(ArdenError):
tool_name: strRaised when a call is blocked by policy or denied by a human (wait mode).
class ApprovalTimeoutError(ArdenError):
action_id: str
timeout: floatRaised in wait mode when no decision is received within max_poll_time.
class ConfigurationError(ArdenError)
Raised when configure() has not been called before a tool call.
For dashboard, API keys, and policy configuration — app.arden.sh. Questions? team@arden.sh