sentinelai-sdk 0.1.5

Drop-in reliability observability for multi-agent AI workflows

sentinelai-sdk

Observability, contract enforcement, and safe replay for multi-agent AI pipelines.

Dashboard: www.agentsentinelai.com/dashboard

---

Install

pip install sentinelai-sdk

Get an API key at the dashboard → ⚙️ Settings → Generate Key. Free, no credit card required.

---

Pick what you need

I want to…	Feature	Dashboard tab
See what my agents are doing — steps, inputs, outputs, latency	Tracing	Traces
Block bad data from reaching the next agent, replay from failure	Contract + Replay	Incidents
Let concurrent agents write shared state without overwriting each other	Shared State	State

These stack — use one, two, or all three.

---

1. Tracing

Zero changes — auto_instrument()

One line. Detects openai, anthropic, langchain and patches them automatically.

import sentinel
sentinel.auto_instrument(api_key="sk_live_...")
sentinel.set_active_run("run_001", "my_pipeline")

# Your existing code — completely unchanged
response = client.chat.completions.create(model="gpt-4o", messages=[...])

Or instrument from the CLI

sentinel instrument pipeline.py

Reads your file, asks what you want to add, writes the instrumented version.

Manual options

# Decorator — 1 line per function
@sentinel.trace_step(name="planner", step_type="llm_call", workflow_name="my_pipeline")
def planner(query): ...

# OpenAI
sentinel.patch_openai(client, workflow_name="my_pipeline")

# Anthropic
sentinel.patch_anthropic(client, workflow_name="my_pipeline")

# LangChain
cb = sentinel.LangChainCallback(workflow_name="my_pipeline")
llm = ChatOpenAI(model="gpt-4o", callbacks=[cb])

---

2. Contract + Replay

Define what one agent must hand to the next. If invalid, Sentinel blocks the downstream agent, creates an incident, and saves a checkpoint for replay.

from sentinel import Sentinel

client = Sentinel(api_key="sk_live_...")

client.register_contract(
    workflow_name="my_pipeline",
    from_step="planner",
    to_step="research",
    schema={
        "type": "object",
        "required": ["destination", "budget", "days"],
        "properties": {
            "destination": {"type": "string"},
            "budget":      {"type": "number"},
            "days":        {"type": "integer"}
        }
    },
    on_fail="block"
)

run = client.start_workflow(workflow_name="my_pipeline", input={"query": "..."})
plan = planner(query)

result = client.record_step(run_id=run["run_id"], step_name="planner", output=plan)
if result["boundary_check"]["result"] == "failed":
    print("Blocked:", result["boundary_check"]["reason"])
    # replay from checkpoint once fixed:
    client.replay(
        run_id=run["run_id"],
        checkpoint_id=result["boundary_check"]["checkpoint_id"],
        patched_output={"destination": "Japan", "budget": 2000, "days": 5}
    )

---

3. Shared State

Safe concurrent writes — no silent overwrites between parallel agents.

# Read
value, version = sentinel.get_state(run_id, "key")

# Write with conflict protection
sentinel.propose_state(run_id, "key", new_value, base_version=version)

# Auto-retry on conflict
sentinel.propose_state_with_retry(run_id, "key", lambda cur: {**(cur or {}), "field": value})

---

Links

• Integration guide
• Runnable examples
• Dashboard