sentinelai-sdk

A control plane for multi-agent AI workflows — tracing, contract enforcement, safe state, and failure replay.

Dashboard: www.agentsentinelai.com/dashboard

---

Install

pip install sentinelai-sdk

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

---

What it does

I want to…	Feature
See every agent step — inputs, outputs, latency, token counts	Tracing
Block bad data from reaching the next agent	Contracts
Replay a failed run from any checkpoint once the bug is fixed	Replay
Let concurrent agents write shared state without overwriting each other	Shared State

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

---

Quickstart

import sentinel

sentinel.init(api_key="sk_live_...")

with sentinel.workflow("my-pipeline") as run:
    with run.step("planner", step_type="llm_call") as step:
        step.set_input({"query": "Plan a trip to Tokyo"})
        result = planner_agent(query)
        step.set_output({"plan": result})

    with run.step("researcher", step_type="tool_call") as step:
        step.set_input(result)
        data = researcher_agent(result)
        step.set_output({"findings": data})

---

Tracing

Option 1 — Workflow context manager (recommended)

import sentinel

sentinel.init(api_key="sk_live_...")

with sentinel.workflow("travel-planner") as run:
    with run.step("plan", step_type="llm_call") as step:
        step.set_input({"query": query})
        output = plan_agent(query)
        step.set_output(output)

Option 2 — Patch OpenAI clients (sync)

Every client.chat.completions.create() call becomes a traced step automatically.

import openai, sentinel

sentinel.init(api_key="sk_live_...")
client = openai.OpenAI(api_key="...")
sentinel.patch_openai(client, workflow_name="my-pipeline")

sentinel.set_active_run("run_001", "my-pipeline")
response = client.chat.completions.create(model="gpt-4o", messages=[...])

Option 3 — Patch AsyncOpenAI clients (async)

import openai, sentinel

sentinel.init(api_key="sk_live_...")
client = openai.AsyncOpenAI(api_key="...")
sentinel.patch_openai_async(client, workflow_name="my-pipeline")

async def main():
    sentinel.set_active_run("run_001", "my-pipeline")
    response = await client.chat.completions.create(model="gpt-4o", messages=[...])

Deep instrumentation — for libraries like gpt-researcher that create their own AsyncOpenAI instances internally, patch at the class level:

import openai.resources.chat.completions as _oai_completions
sentinel.patch_openai_async(_oai_completions.AsyncCompletions)
# Now every AsyncOpenAI client anywhere in the process is traced

Option 4 — LangChain callback

from sentinel import LangChainCallback
from langchain_openai import ChatOpenAI

sentinel.init(api_key="sk_live_...")
cb = LangChainCallback(workflow_name="my-pipeline")
llm = ChatOpenAI(model="gpt-4o", callbacks=[cb])

Option 5 — Decorator

@sentinel.trace_step(name="planner", step_type="llm_call", workflow_name="my-pipeline")
def planner(query):
    return llm.invoke(query)

---

Contracts (handoff validation)

Define what one agent must pass to the next. If the payload fails validation, Sentinel raises ContractViolationError, marks the run as blocked in the dashboard, and saves a checkpoint for replay.

import sentinel
from sentinel import ContractViolationError

sentinel.init(api_key="sk_live_...")

sentinel.register_contract(
    agent="researcher",
    accepts={
        "destination": {"type": "string",  "required": True, "min_length": 1},
        "budget":      {"type": "number",  "required": True, "min": 100},
        "days":        {"type": "number",  "required": True, "min": 1, "max": 30},
    },
)

with sentinel.workflow("travel-planner") as run:
    with run.step("planner", step_type="llm_call") as step:
        plan = planner_agent(query)
        step.set_output(plan)

    # Validate handoff before the next agent runs
    sentinel.handoff(
        from_agent="planner",
        to_agent="researcher",
        payload=plan,
        run_id=run.run_id,
    )

    with run.step("researcher", step_type="tool_call") as step:
        research = researcher_agent(plan)
        step.set_output(research)

Field spec options

{"type": "string",  "required": True, "min_length": 1}
{"type": "number",  "required": True, "min": 0, "max": 100}
{"type": "boolean", "required": True}
{"type": "array",   "required": False}

---

Shared State

Safe concurrent writes — no silent overwrites when agents run in parallel.

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

# Write with conflict protection (raises ConflictError if stale)
sentinel.propose_state(run_id, "research_results", new_value, base_version=version)

# Auto-retry on conflict (merge function receives current value)
sentinel.propose_state_with_retry(
    run_id, "research_results",
    lambda cur: {**(cur or {}), "hotels": hotel_list}
)

---

gpt-researcher example

Full deep instrumentation — every internal LLM call becomes a visible step:

import asyncio, os
import sentinel
from sentinel import ContractViolationError
import openai.resources.chat.completions as _oai_completions

sentinel.init(api_key="sk_live_...")

# Patch at the class level to catch gpt-researcher's internal AsyncOpenAI clients
sentinel.patch_openai_async(_oai_completions.AsyncCompletions)

sentinel.register_contract(
    agent="write_report",
    accepts={
        "source_count": {"type": "number",  "required": True, "min": 1},
        "has_context":  {"type": "boolean", "required": True},
        "query":        {"type": "string",  "required": True, "min_length": 1},
    },
)

async def run_research(query: str) -> str:
    from gpt_researcher import GPTResearcher
    researcher = GPTResearcher(query=query, report_type="research_report", verbose=False)

    with sentinel.workflow("gpt-researcher") as run:
        sentinel.set_active_run(run.run_id, "gpt-researcher")

        with run.step("conduct_research", step_type="tool_call") as step:
            step.set_input({"query": query})
            await researcher.conduct_research()
            sources = researcher.get_source_urls()
            context = researcher.get_research_context()
            step.set_output({"source_count": len(sources), "sources": sources[:5]})

        sentinel.handoff(
            from_agent="conduct_research",
            to_agent="write_report",
            payload={"source_count": len(sources), "has_context": bool(context), "query": query},
            run_id=run.run_id,
        )

        with run.step("write_report", step_type="llm_call") as step:
            step.set_input({"source_count": len(sources), "query": query})
            report = await researcher.write_report()
            step.set_output({"word_count": len(report.split())})

    return report

asyncio.run(run_research("What is the impact of AI agents on software engineering in 2025?"))

Run with DuckDuckGo (no API key needed):

pip install gpt-researcher sentinelai-sdk ddgs
RETRIEVER=duckduckgo OPENAI_API_KEY=sk-... python your_script.py

Each run produces ~5 nested steps in the dashboard:

• openai/gpt-4.1 — agent selection
• conduct_research — phase wrapper
• openai/o4-mini — report generation
• write_report — phase wrapper
• openai/gpt-4.1 — TOC / introduction / conclusion

---

Links

• Dashboard
• GitHub
• Examples