Warden

Open source observability for AI agent trajectories and tool execution failures.

Warden is a full-stack platform — local SDKs, an ingest API, and a dashboard — that makes silent agent failures loud. The same MIT codebase runs in-process, self-hosted, or as managed Warden Cloud.

• Assert locally, before a tool body runs: missing prerequisites, parallel sequence bypasses, duplicate overlapping calls, or runaway retry loops.
• Trace every agent run and tool call with OpenTelemetry and OpenInference semantic conventions (agent.name, tool.name, tool.parameters, tool.output, …).
• Export AgentTracePayload documents to the ingest API for replay in the dashboard.

Warden Cloud vs self-host vs SDK-only

There is no private folder in this repo. GitHub is all-or-nothing, so we follow the Langfuse model: one public tree, one license, Cloud is the hosted product.

	What you run	When to use it
Warden Cloud	The SDK, pointed at our ingest	Fastest path. We operate ClickHouse, the API, and setup. Sign up and send traces.
Self-host	server/ + web/ on your infra	Data stays on your machines. Same code as Cloud. See those READMEs (local uvicorn / Render + Vercel).
SDK only	Python warden-agent or @warden/sdk	Assertions and traces in-process. Leave endpoint unset and exports go nowhere.

Repository

Path	Role
src/	Python SDK (warden-agent)
ts/	TypeScript SDK (@warden/sdk)
server/	FastAPI ingest + query API
web/	Next.js Cloud frontend
contract/	Wire schema (warden.json)
examples/	No-LLM harnesses

Control plane

The Vercel-ready Next.js frontend lives in web/. It provides Clerk authentication, a landing and pricing page, and a setup flow that issues a project API key and walks through SDK install plus a first request. The trace dashboard is not in this UI yet. See web/README.md for local setup and deployment variables.

Wire contract

The payload schema (AgentTracePayload / AgentStep) is defined once in the server's Pydantic schemas (server/src/warden_server/schemas.py) and derived into a language-neutral artifact at contract/warden.json by server/scripts/emit_contract.py. Consumers are generated or verified from that artifact — nothing is hand-synced:

• TypeScript SDK — ts/src/generated/contract.ts is generated from the artifact with pnpm codegen (ts/).
• Python SDK — tests/test_contract.py validates a real SDK export against the artifact with jsonschema.
• Server — server/tests/test_contract.py fails if contract/warden.json goes stale relative to schemas.py.

Change the contract like this:

# edit server/src/warden_server/schemas.py, then:
python server/scripts/emit_contract.py   # regenerates contract/warden.json
(cd ts && pnpm codegen)                  # regenerates ts/src/generated/contract.ts

Run the full suite (uv run ruff check ., uv run pytest at the root, and pnpm typecheck && pnpm test && pnpm build in ts/).

Installation

Using uv (recommended):

uv sync            # creates .venv, installs the package + dev group
uv run pytest      # run the test suite
uv run ruff check .  # lint

Or with plain pip:

pip install -e ".[sdk]"   # add `[dev]`-style deps with whatever tool you prefer

Quickstart

from warden import Warden

sentry = Warden(
    endpoint="https://ingest.example.com/v1/traces",
    api_key=os.environ["WARDEN_API_KEY"],
)

# --- decorate atomic actions --------------------------------------------
@sentry.track_tool
def verify_identity(user_id: str, password: str) -> bool:
    return user_id == "root" and password == "hunter2"

# Refunds must only ever run after an identity check in this trajectory.
@sentry.track_tool(required_sequence=["verify_identity"])
def process_refund(order_id: str, amount: float) -> str:
    return f"refunded ${amount:.2f} for {order_id}"

# LLM loops that retry with identical args trip this guard after 3 tries.
@sentry.track_tool(max_repeat_count=3)
def query_knowledge_base(question: str) -> str:
    ...

# --- decorate top-level agents ------------------------------------------
@sentry.track_agent
def refund_agent(order_id: str, session_id: str) -> str:
    ok = verify_identity("root", "hunter2")
    if not ok:
        return "auth failed"
    return process_refund(order_id, 19.99)

Every call to refund_agent() produces one AgentTracePayload:

{
  "session_id": "9f4c...",
  "status": "SUCCESS",
  "total_duration_ms": 128.44,
  "steps": [
    {
      "type": "agent",
      "name": "refund_agent",
      "inputs": {},
      "outputs": "refunded $19.99 for ord_123",
      "latency_ms": 1.31
    },
    {
      "type": "tool",
      "name": "verify_identity",
      "inputs": {"user_id": "root", "password": "***"},
      "outputs": true,
      "latency_ms": 0.87
    },
    {
      "type": "tool",
      "name": "process_refund",
      "inputs": {"order_id": "ord_123", "amount": 19.99},
      "outputs": "refunded $19.99 for ord_123",
      "latency_ms": 2.02
    }
  ]
}

Traffic-Light Statuses

The status field on a payload tells you how the agent run finished:

Status	Meaning
SUCCESS	Agent returned normally. Handled assertion steps may still be present.
CRASHED	An unhandled exception escaped the agent function.
ASSERTION_FAILED	An unhandled SequenceViolationError, ParallelInvocationError, or InfiniteLoopError escaped.

Local Assertion Engine

Warden's killer feature. Rules run before a tool body is invoked, using the trajectory captured so far (warden.assertions). The TypeScript package @warden/sdk is a native in-process port of the same rules (Sentry-style: npm i does not start Python).

Required sequence (including parallel harnesses)

A prerequisite must have completed successfully. If a harness launches verify_identity and process_refund in the same asyncio.gather / Promise.all turn, process_refund is rejected — in-flight does not count.

from warden import SequenceViolationError

try:
    process_refund("ord_123", 19.99)
except SequenceViolationError as exc:
    print(exc.missing)  # ["verify_identity"]

Duplicate overlapping calls

@sentry.track_tool(allow_parallel=False)
def process_refund(order_id: str) -> str: ...

A second call that starts before the first finishes raises ParallelInvocationError. Independent fan-out (two searches) stays allowed by default (allow_parallel=True).

Infinite-loop prevention

from warden import InfiniteLoopError

@sentry.track_tool(max_repeat_count=3)          # default is 3
def call_llm(prompt: str) -> str: ...

for _ in range(4):
    call_llm("same prompt")   # 4th consecutive identical call raises
# -> InfiniteLoopError: Tool 'call_llm' invoked 4 consecutive times with
#    identical arguments (max_repeat_count=3): {'prompt': 'same prompt'}

The counter is per (tool, arguments-hash) and resets as soon as a tool is called with different arguments. Set max_repeat_count=None to disable.

Assertion failure policy

Low-level track_tool keeps hard-fail behavior by default. Framework adapters default to tool_error: the prohibited tool body stays blocked, while the host receives a JSON-safe result and can continue its agent loop.

@sentry.track_tool(
    required_sequence=["verify_identity"],
    assertion_failure_mode="tool_error",  # or "raise"
)
def process_refund(order_id: str): ...

The TypeScript equivalents are assertionFailureMode and assertionFailureFormatter. Python uses assertion_failure_formatter. Handled failures retain an ASSERTION_FAILED step with assertion.handled=true, but an otherwise completed trajectory is SUCCESS. Set the mode to raise for security-critical policies that must propagate to the host. If a framework converts all tool exceptions into model feedback, configure its own fatal-error policy as well.

Hermes built-in hooks are the exception: Hermes swallows hook exceptions, so install() only accepts tool_error and always returns its safe block directive. Use guard_handler(..., assertion_failure_mode="raise") when you own the Hermes tool handler and need propagation.

Generic wraps

Chat Completions / Responses registries and Vercel tool({ execute }) objects use the in-process engine without a host SDK:

from warden.adapters import wrap_openai_tools, invoke_openai_tool_calls

registry = wrap_openai_tools(sentry, TOOL_REGISTRY, {
    "process_refund": {"required_sequence": ["verify_identity"], "allow_parallel": False},
})
results = await invoke_openai_tool_calls(registry, message.tool_calls)
# Blocked calls are returned as {"error": {"type": "warden_assertion", ...}}.
# Pass assertion_failure_mode="raise" for a hard failure.

import { trackVercelTool } from "@warden/sdk/vercel";
import { wrapOpenAITools } from "@warden/sdk/openai";

export const refund = tool(trackVercelTool(sentry, {
  description: "Refund an order",
  inputSchema: z.object({ order_id: z.string() }),
  async execute({ order_id }) { return `refunded ${order_id}`; },
}, { name: "process_refund", requiredSequence: ["verify_identity"], allowParallel: false }));

OpenAI Agents, Hermes, Eve, and OpenClaw have host adapters (see Framework adapters). wrap_callable is the shared primitive those hosts sit on.

See examples/parallel_tools.py and ts/README.md.

Redaction & filtering

By default, values of keys like password, token, secret, api_key, authorization, cookie and x-api-key are replaced with "***" in the captured inputs:

@sentry.track_tool(redact={"password", "ssn"})     # customize
@sentry.track_tool(redact=())                      # disable redaction

@sentry.track_tool(include=["username"])           # record only these params

Sessions & context

Every trajectory is scoped to a session_id, passed to your agent function as a keyword argument (a UUID is used when omitted). Tools automatically join the enclosing agent's session; tools called outside of an agent share a per-thread context so assertion state still accumulates.

from warden import get_current_context, get_current_session_id

assert get_current_session_id() == my_session_id
ctx = get_current_context()
ctx.record_metric("tokens_used", 1234)
print(ctx.executed_tools)          # ["verify_identity", "process_refund"]
print(ctx.memory)                  # shared mutable dict for the session

Framework-owned run loops (OpenAI Agents Runner.run, Hermes, OpenClaw) can bind tools to a session without decorating the host:

with sentry.session_context("sess-001"):
    verify_identity("root", "hunter2")
    process_refund("ord_123", 19.99)

sentry.start_session("sess-001", agent_name="refund_agent")
sentry.end_session("sess-001")   # flush + drop

Framework adapters

Warden wraps the tool surface of the agent frameworks we integrate with. None of these packages are required at install time — adapters duck-type the host APIs. Install the host SDK yourself when you want the real runtime (pip install openai-agents, Hermes from Nous Research, OpenClaw from npm).

Those wrappers are the assertion path: they see tool calls, not the host's LLM round-trips. For the server/web dashboard, use an auto-instrumentation library so LLM inputs, tool names, latency, and outputs arrive as OpenTelemetry GenAI / OpenInference spans. See Auto-instrumentation.

OpenAI Agents SDK (Python)

from warden import Warden
from warden.openai_agents import track_openai_tool, wrap_function_tool

sentry = Warden(endpoint="https://ingest.example.com/v1/traces")

@track_openai_tool(sentry, required_sequence=["verify_identity"])
def process_refund(order_id: str) -> str:
    return f"refunded {order_id}"

# Already constructed FunctionTool:
wrap_function_tool(sentry, existing_tool, required_sequence=["verify_identity"])

@sentry.track_agent
async def run(prompt: str, session_id: str) -> str:
    result = await Runner.run(agent, prompt, context={"session_id": session_id})
    return result.final_output

Put session_id on the run context so tools join the same trajectory. uv run python examples/openai_agents_demo.py is the no-LLM harness.

The TypeScript equivalent is trackOpenAITool from @warden/sdk/openai-agents.

Hermes Agent

Install Warden as a plugin so built-in tools (terminal, etc.) are guarded. pre_tool_call returns Hermes' {"action": "block", "message": ...} directive on sequence/loop violations so the tool never runs:

from warden.hermes import install, guard_handler

def register(ctx):
    install(
        ctx,
        sentry,
        rules={"terminal": {"required_sequence": ["authenticate"]}},
    )
    ctx.register_tool(
        name="issue_refund",
        toolset="warden-demo",
        schema={...},
        handler=guard_handler(sentry, "issue_refund", issue_refund,
                              required_sequence=["authenticate"]),
    )

uv run python examples/hermes_plugin_demo.py simulates the plugin host.

OpenClaw

TypeScript: wrap tools before api.registerTool / defineToolPlugin:

import { trackOpenClawTool } from "@warden/sdk/openclaw";

api.registerTool(trackOpenClawTool(sentry, {
  name: "process_refund",
  async execute(_id, params, ctx) { return { refunded: true }; },
}, { requiredSequence: ["authenticate"] }));

OpenClaw's own sandbox (agents.defaults.sandbox.backend: docker) is gateway config; Warden traces the call regardless of where the host executes it.

Eve (TypeScript)

Same composition as OpenClaw: wrap, then hand to the host (defineTool(trackEveTool(...)) from @warden/sdk/eve). Real Eve runs also register createWardenSpanExporter because Eve's native telemetry is OpenTelemetry spans and Eve owns the run loop.

Auto-instrumentation (server / web)

Observe and enforce are separate paths (the Langfuse lesson):

Path	What it does	When to use it
Enforce	track_tool / harness wrappers run assertions before the tool body	You own the tool execute / handler
Observe	OpenInference / GenAI / Eve OTel spans fold into AgentTracePayload	The framework owns the run loop (Eve, ADK, Claude Agent SDK) or you want LLM I/O

Use them together when you can. On Eve / ADK / Claude Agent SDK, trackAgent cannot wrap the host loop — register an OTel exporter (or POST /v1/otel/v1/traces) for the trajectory tree, and keep wrappers only on tools you can intercept. Assertion state from trackAgent is not shared with the OTel path.

An auto-instrumentation wrapper — openinference-instrumentation-openai-agents, pydantic-ai[logfire], Eve otelIntegration, OpenClaw diagnostics-otel — wraps the host SDK, extracts LLM inputs, tool names, latency, and outputs, and emits OpenTelemetry GenAI / OpenInference spans. Warden folds those spans into the same AgentTracePayload the simple SDK posts to /v1/traces, so the dashboard can show the full trajectory — including span_id / parent_id nesting, per-turn trace_id, and optional user_id.

Instrumentation scopes such as Eve durable workflow and @vercel/otel/fetch are dropped so the GenAI / Eve parent is not orphaned by HTTP noise.

In-process (Python)

from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor
from warden.otel import WardenSpanExporter

provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(
    WardenSpanExporter(
        endpoint="https://api.example.com/v1/traces",
        api_key="wk_...",
    ),
))
OpenAIAgentsInstrumentor().instrument()

pydantic-ai[logfire] is the same idea: Logfire's instrumentor emits GenAI spans. Point a WardenSpanExporter (or OTLP below) at Warden instead of Logfire Cloud. Host instrumentors are optional — they are not Warden install-time dependencies.

uv run python examples/otel_auto_instrument.py folds a fake OpenInference batch with no live model.

OTLP/HTTP (any language)

The OTLP exporter appends /v1/traces to OTEL_EXPORTER_OTLP_ENDPOINT:

export OTEL_EXPORTER_OTLP_ENDPOINT=https://api.example.com/v1/otel
export OTEL_EXPORTER_OTLP_PROTOCOL=http/json
export OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer wk_...

JSON only (415 for protobuf). Gzip Content-Encoding is accepted. Auth is the same project API key as POST /v1/traces. OpenClaw diagnostics-otel and any OTel SDK can use this path; Eve TypeScript still prefers createWardenSpanExporter in-process because Eve skips custom span processors.

Configuration

Warden(...) kwarg	Default	Description
endpoint	None	Ingestion URL. If unset, traces go to /dev/null.
api_key	None	Sent as Authorization: Bearer <key>.
headers	None	Extra HTTP headers for the exporter.
exporter	None	Custom TraceExporter (e.g. OTLP, S3, in-memory).
auto_flush	True	Export the payload when each agent finishes.
timeout	10.0	HTTP exporter timeout (seconds).

OpenTelemetry

The SDK mirrors each Warden step to an outgoing OTel span (agent.* / tool.*) when an opentelemetry-sdk provider is configured.

The reverse path — auto-instrumentation into Warden — is Auto-instrumentation: GenAI / OpenInference spans are folded into AgentTracePayload by warden.otel.WardenSpanExporter or POST /v1/otel/v1/traces.

Development

uv sync                        # install deps + dev group into .venv
uv run pytest                  # run the test suite (incl. example harnesses)
uv run ruff check .            # lint
uv run python examples/refund_agent.py
uv run python examples/live_agent_demo.py   # live demo, no API key needed
uv run python examples/openai_function_calling_agent.py  # offline without OPENAI_API_KEY
uv run python examples/openai_agents_demo.py  # OpenAI Agents SDK adapter, no LLM
uv run python examples/hermes_plugin_demo.py  # Hermes plugin hooks, no hermes-agent pkg
uv run python examples/otel_auto_instrument.py  # OpenInference/GenAI fold, no live model
uv run python examples/parallel_tools.py    # gather() sequence + duplicate guards
# Docker sandbox (as in CI):
WARDEN_SANDBOX_BACKEND=docker uv run pytest tests/test_harness_sandbox.py tests/test_framework_sandbox.py

The Python version is pinned in .python-version and dependencies are locked in uv.lock for reproducible builds. Commit both.

License

This repository is MIT licensed — SDKs, server, and web. There is no separate commercial source tree today. Managed Warden Cloud (hosted ingest, retention, support) is how the project is funded.