Sentrial Python SDK

Observability for AI agents — with automated root cause analysis and fix suggestions.

Track LLM calls, tool executions, costs, and latency. When things go wrong, Sentrial tells you WHY and suggests how to fix it.

Why Sentrial?

Installation

# Core (works with any framework)
pip install sentrial

# With specific providers
pip install sentrial[openai]      # OpenAI auto-tracking
pip install sentrial[anthropic]   # Anthropic auto-tracking
pip install sentrial[google]      # Google/Gemini auto-tracking
pip install sentrial[langchain]   # LangChain callback handler
pip install sentrial[openai-agents]  # OpenAI Agents SDK trace processor
pip install sentrial[google-adk]  # Google ADK Plugin / callbacks
pip install sentrial[otel]        # OpenTelemetry integration

# Everything
pip install sentrial[all]

Breaking changes

0.7.0

• Transport is now OpenTelemetry (OTLP). Telemetry ships over /api/otel/v1/traces; the legacy /api/sdk/* session-ingestion endpoints have been removed. The OTLP ingest route requires the API key to carry both sessions:write and events:write scopes — dashboard-generated keys include both, but a manually-scoped sessions:write-only key will silently drop data. Add events:write to the key (see Transport).
• PII config change. PiiConfig.enhanced_detection was removed and the default redacted-field set changed. Review your PiiConfig(...) if you relied on the previous defaults.

Quick Start (30 seconds)

Option 1: Wrap your LLM client (Recommended)

OpenAI:

from sentrial import wrap_openai, configure, begin
from openai import OpenAI

configure(api_key="sentrial_live_xxx")
client = wrap_openai(OpenAI())  # ← That's it!

with begin(user_id="user_123", event="chat", input=user_message) as session:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_message}]
    )
    # ✔️ Automatically tracked: model, tokens, cost, latency
    session.set_output(response.choices[0].message.content)

Anthropic:

from sentrial import wrap_anthropic, configure, begin
from anthropic import Anthropic

configure(api_key="sentrial_live_xxx")
client = wrap_anthropic(Anthropic())

with begin(user_id="user_123", event="chat", input=user_message) as session:
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_message}]
    )
    session.set_output("\n".join(block.text for block in response.content if block.type == "text"))

Google Gemini:

from sentrial import wrap_google, configure, begin
import google.generativeai as genai

configure(api_key="sentrial_live_xxx")
genai.configure(api_key="your_google_key")
model = wrap_google(genai.GenerativeModel("gemini-2.5-pro"))

with begin(user_id="user_123", event="chat", input=prompt) as session:
    response = model.generate_content(prompt)
    session.set_output(response.text)

Option 2: Use decorators (Simplest)

from sentrial import tool, session, configure

configure(api_key="sentrial_live_xxx")

@tool("search_web")
def search_web(query: str) -> dict:
    """Tool calls are automatically tracked"""
    return {"results": [...]}

@tool("get_weather")
def get_weather(city: str) -> dict:
    return {"temp": 72, "condition": "sunny"}

@session("my-agent")
def run_agent(user_id: str, message: str) -> str:
    """Session boundary - tracks input/output automatically"""
    results = search_web(message)
    weather = get_weather("San Francisco")
    return f"Found {len(results)} results. Weather: {weather}"

# Run it - everything is tracked!
run_agent(user_id="user_123", message="What's happening today?")

Option 3: FastAPI / Async

from fastapi import FastAPI
from sentrial import AsyncSentrialClient
from contextlib import asynccontextmanager

sentrial = AsyncSentrialClient(
    api_key="sentrial_live_xxx",
    api_url="http://localhost:3001"
)

@asynccontextmanager
async def lifespan(app):
    yield
    await sentrial.close()

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
async def chat(request: ChatRequest):
    async with await sentrial.begin(
        user_id=request.user_id,
        event="chat",
        input=request.message,
    ) as session:
        # Your agent logic here
        result = await call_llm(request.message)

        await session.track_tool_call(
            tool_name="llm_call",
            tool_input={"prompt": request.message},
            tool_output={"response": result}
        )

        session.set_output(result)
        return {"response": result}

Option 4: LangChain (ALL versions!)

Works with LangChain 0.x AND 1.x - our SDK handles version differences automatically.

from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from sentrial import SentrialClient, create_agent_with_sentrial

client = SentrialClient(api_key="sentrial_live_xxx")
llm = ChatOpenAI(model="gpt-4o")

@tool
def my_tool(query: str) -> str:
    """Search for information."""
    return f"Results for: {query}"

# create_agent_with_sentrial handles ALL LangChain versions!
agent = create_agent_with_sentrial(
    llm=llm,
    tools=[my_tool],
    client=client,
    agent_name="support-agent",
    user_id="user_123",
)

result = agent("How do I reset my password?")
print(result)

from langgraph.prebuilt import create_react_agent
from sentrial.langchain import SentrialCallbackHandler

agent = create_react_agent(llm, tools)
handler = SentrialCallbackHandler(client, session_id)

result = agent.invoke(
    {"messages": [("user", query)]},
    config={"callbacks": [handler]}
)

from langchain.agents import AgentExecutor, create_openai_tools_agent
from sentrial.langchain import SentrialCallbackHandler

agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools)
handler = SentrialCallbackHandler(client, session_id)

result = executor.invoke(
    {"input": query},
    config={"callbacks": [handler]}
)

Option 5: CrewAI

from crewai import Agent, Task, Crew
from sentrial import SentrialClient
from sentrial.crewai import SentrialCrewHandler

client = SentrialClient(api_key="sentrial_live_xxx")
session_id = client.create_session(name="Research Crew", agent_name="crewai-research", user_id="user_123")
handler = SentrialCrewHandler(client, session_id)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    step_callback=handler.on_step,
    task_callback=handler.on_task_complete,
)

result = crew.kickoff(inputs={"topic": "AI agents"})
handler.finish(success=True, output=str(result))

Option 6: AutoGen

from autogen_agentchat.teams import RoundRobinGroupChat
from sentrial import SentrialClient
from sentrial.autogen import SentrialAutogenHandler

client = SentrialClient(api_key="sentrial_live_xxx")
session_id = client.create_session(name="Multi-Agent Chat", agent_name="autogen-team", user_id="user_123")
handler = SentrialAutogenHandler(client, session_id)

team = RoundRobinGroupChat([agent1, agent2])
result = await handler.run_team(team, task="Analyze this data...")
handler.finish(success=True)

Option 7: OpenAI Agents SDK

One call registers a trace processor with the Agents SDK — every run becomes a Sentrial session with LLM events, tool calls, handoffs, and guardrails. Don't combine with wrap_openai() on the same run (LLM calls would be double-counted).

import sentrial
from agents import Agent, Runner
from sentrial.openai_agents import instrument_openai_agents

sentrial.configure(api_key="sentrial_live_xxx")
instrument_openai_agents(user_id="user_123")

agent = Agent(name="Assistant", instructions="You are a helpful assistant")
result = Runner.run_sync(agent, "Write a haiku about recursion.")
# The run is now a Sentrial session with LLM events, tool calls, handoffs.

Requires pip install 'sentrial[openai-agents]'. See the OpenAI Agents SDK integration docs for configuration options, the run→session mapping, and batching/flush guidance.

Option 8: Google ADK (Agent Development Kit)

Register one Plugin on your ADK Runner — every run becomes a Sentrial session with model events, tool calls, tokens, and cost. Don't combine with wrap_google() on the same run (LLM calls would be double-counted).

import sentrial
from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from sentrial.google_adk import SentrialAdkPlugin

sentrial.configure(api_key="sentrial_live_xxx")

agent = LlmAgent(name="assistant", model="gemini-2.0-flash")
runner = InMemoryRunner(agent=agent, plugins=[SentrialAdkPlugin()])
# Every runner.run_async(...) is now a Sentrial session.

Requires pip install 'sentrial[google-adk]'. A per-agent sentrial_adk_callbacks() helper is also available. See the Google ADK integration docs for configuration options, the run→session mapping, and batching guidance.

Option 9: OpenTelemetry (Enterprise)

Works with any OTel-instrumented framework (Vercel AI SDK, LangChain, etc.):

from sentrial.otel import setup_otel_tracing

# One line setup
setup_otel_tracing(
    api_key="sentrial_live_xxx",
    project="my-ai-app"
)

# Now ANY OTel-instrumented library sends traces to Sentrial!
# Works with: opentelemetry-instrumentation-openai, traceloop, etc.

Or add to existing OTel setup:

from opentelemetry.sdk.trace import TracerProvider
from sentrial.otel import SentrialSpanProcessor

provider = TracerProvider()
provider.add_span_processor(SentrialSpanProcessor(
    api_key="sentrial_live_xxx",
    project="my-ai-app"
))

What Gets Tracked

Environment Variables

export SENTRIAL_API_KEY=sentrial_live_xxx
export SENTRIAL_API_URL=https://api.sentrial.com  # or http://localhost:3001 for local

Then just:

from sentrial import configure
configure()  # Uses env vars automatically

Optional agent reports

interaction.report_diagnostic({
    "reportId": "claim-context-1",
    "category": "missing_context",
    "state": "reported",
    "description": "The policy number needed for this claim is missing.",
})
# Later, reuse reportId with state="recovered" and a new description.

Use await interaction.report_diagnostic(...) with the async SDK. The client methods are client.report_diagnostic(session_id, report) and its async equivalent. Categories are missing_context, tool_problem, missing_capability, and uncertain_completion. The dictionary keys match the shared wire contract in both languages.

Reports and claimed recovery remain agent assertions, not confirmed failures, human labels or business outcomes. Each observation is retained separately and never changes session success or invokes a model. Use an opaque report ID of 1–128 letters, digits, ., _, :, or -, starting with a letter/digit. Descriptions are bounded to 4,096 UTF-8 bytes. Optional relatedSpanIds accepts up to eight distinct nonzero lowercase 16-digit OTel span IDs in this trace; references remain unverified and are not followed automatically. Normal metadata redaction, batching, retry and fail-silent behavior apply. A return value is not a durable-delivery receipt; inspect delivery diagnostics and retained events. Degraded/finished interactions are no-ops.

The updated collector and UI display Agent report and Reported recovered. Older collectors retain generic metadata but cannot provide the validated view. Existing signal evidence selectors do not use self-reports as authoritative evidence; incremental recall and review burden remain evaluation work.

Transport

The SDK sends telemetry over OpenTelemetry (OTLP) to /api/otel/v1/traces — this is the sole ingestion path (the legacy /api/sdk/* session endpoints have been removed). The OTLP ingest route requires the API key to carry both sessions:write and events:write scopes — keys generated in the dashboard include both automatically. If you use a manually-scoped key that only has sessions:write, session creation will 403 and tracking is silently dropped; add events:write to the key.

Events and completion attach to an in-memory span the client holds from create_session() / begin(), so they only apply to sessions created on the same client instance. If you obtain a session id elsewhere — a session created out-of-band, shared across processes, or tracked on a different SentrialClient — its events and finalization are skipped (the SDK logs a one-time warning). Use the same client instance for the full begin→track→finish flow.

API Reference

Configuration

from sentrial import configure

configure(
    api_key="sentrial_live_xxx",      # Or SENTRIAL_API_KEY env var
    api_url="https://api.sentrial.com" # Or SENTRIAL_API_URL env var
)

LLM Wrappers

from sentrial import wrap_openai, wrap_anthropic, wrap_google

# Wrap once, use everywhere
openai_client = wrap_openai(OpenAI())
anthropic_client = wrap_anthropic(Anthropic())
google_model = wrap_google(genai.GenerativeModel("gemini-2.5-pro"))

Context Manager

from sentrial import begin

with begin(
    user_id="user_123",           # Required: for user analytics
    event="agent_name",           # Required: groups sessions
    input="user message",         # Optional: captured as session input
) as session:
    # Track tool calls
    session.track_tool_call(
        tool_name="search",
        tool_input={"query": "..."},
        tool_output={"results": [...]}
    )

    # Track decisions/reasoning
    session.track_decision(
        reasoning="Choosing to search because...",
        confidence=0.9
    )

    # Set output
    session.set_output("Final response to user")

Decorators

from sentrial import tool, session

@tool("tool_name")
def my_tool(arg1: str) -> dict:
    """Automatically tracked when called within a session"""
    return {"result": "..."}

@session("agent_name")
def my_agent(user_id: str, message: str) -> str:
    """Creates session, tracks input/output, handles errors"""
    return my_tool(message)

Async Client

from sentrial import AsyncSentrialClient

client = AsyncSentrialClient(api_key="...")

async with await client.begin(user_id="...", event="...") as session:
    await session.track_tool_call(...)
    session.set_output("...")

# Don't forget to close when done
await client.close()

Framework Compatibility

Examples

See the repo-root examples/ directory (and its README) for the framework-specific catalog — OpenAI/Anthropic/Google wrappers, decorators, FastAPI, LangChain, CrewAI, AutoGen, and OTel examples. For an end-to-end walkthrough, see packages/python-sdk/examples/performance_monitoring.py.

Dashboard Features

After tracking, view in the web dashboard:

• Sessions: See all agent runs with input/output
• Events: Drill into tool calls, LLM decisions, errors
• Users: Track daily active users, session counts
• Agents: Compare performance across agents
• Analytics: Cost trends, latency percentiles, token usage

Support

• 📚 Documentation
• 💬 Discord
• 📧 Email
• 🐛 GitHub Issues

SDK feature parity (TypeScript ↔ Python)

Both SDKs share the same core API (begin/finish, tool & LLM tracking, production experiments, provider wrappers). Framework integrations currently differ:

Porting the missing TypeScript integrations (Langfuse, Claude multi-turn) is planned; new TS integrations must keep V1 API response shapes backward-compatible.

License

MIT License - see LICENSE for details.

Delivery diagnostics and worker handoffs

Large tool content (opt-in preview)

The native sync and async clients can upload serialized tool inputs/outputs above 64 KiB directly to private storage and attach small references to the tool span. The configured PII policy runs before hashing or upload; redaction_revision labels the policy and does not redact anything itself.

from sentrial import SentrialClient, LargeToolContentConfig

client = SentrialClient(
    api_key="your-ingestion-key",
    pii={"enabled": True},
    large_tool_content=LargeToolContentConfig(
        enabled=True,
        redaction_revision="claims-redaction-v1",
        timeout_seconds=20,
    ),
)

AsyncSentrialClient, configure() and configure_async() accept the same configuration (or a dict with those keys). This is disabled by default and requires separately enabled server storage and workspace quotas. It applies to named native tool_call inputs/outputs, including calls recorded through native integrations. LLM messages, arbitrary external OpenTelemetry spans and metadata/state snapshots retain their existing handling.

Each field keeps a prefix of at most 4 KiB of UTF-8 source bytes inline. Objects are limited to 16 MiB, with two concurrent uploads per client, at most 24 requests per field and bounded control responses. Lost PUT acknowledgements are checked against retained parts before retry; create/complete ambiguity is reconciled rather than blindly replayed. Oversize, unavailable, cancelled and capacity-limited uploads retain an excerpt and explicit missing-content marker. Small fields and clients without the opt-in keep their existing inline behavior. Resume is in memory; restarting the SDK does not resume its upload. Dedicated upload transports do not inherit HTTP cookies, proxy environment credentials or netrc authentication.

The per-field deadline is 1–60 seconds (20 by default), excluding normal trace delivery. Synchronous OS networking may continue after cancellation: at most two daemon workers can remain, and they retain their capacity slots until they actually exit. The caller receives a timeout/missing outcome; late work cannot convert it to success. Async cancellation closes the request path, and cancellation initiated by the caller still propagates. No SDK cancellation is proof that a storage provider has stopped an already accepted write.

complete_session() and flush() wait for captures already started before draining their telemetry. shutdown() and close() cancel outstanding uploads and drain fallback events; flush first when you want to wait for uploads. The async methods use await. Finish only after the application work belonging to that session is done, and do not record more events after finishing. Abrupt termination cannot guarantee delivery.

get_content_upload_diagnostics() returns attempted/submitted/missing/active/pending-event counters, or None when disabled. Submitted content still needs server verification and separate detector inspection. A UI preview does not establish detector coverage. Implementation and rollout limits.

Request and delivery diagnostics

For request latency and response reports, read client.get_transport_diagnostics() (also available without batching). It separates logical requests, retries, in-flight attempts, response outcomes, reported span acceptance and fixed server timing stages. It retains lifetime aggregates plus 32 recent completed attempts, with no payloads or identifiers. Missing values stay unknown; a successful HTTP response is not proof of unique durable writes. Usage and measurement limits.

Call finish after the work belonging to a session has completed, then explicitly flush/close before process exit. Native SDK spans carry source sequence numbers and a final count, so missing events can be detected independently of server arrival order. This proves delivery of emitted spans, not that every application action was instrumented. Queues remain in memory and can drop events at capacity; abrupt termination cannot guarantee a flush.

Read client.get_delivery_diagnostics() for payload-free batcher counters (None when no batcher exists). failed means delivery was not confirmed; cancelled is a subset of failures. A cancelled send can have reached the server. The uncertain send and untouched tail are retried in order within the queue limit; delivery can therefore be duplicated. Counters count attempts, so delivered + failed can exceed enqueued after a retry.

from sentrial import action_metadata, resume_handoff_context

parent = client.begin(user_id='example', event='workflow',
                      metadata=action_metadata('order:42'))
carrier = client.handoff(parent.session_id, 'order:42') if parent.session_id else None
# Pass carrier through your job queue, then in the worker:
child = worker_client.begin(user_id='example', event='worker',
                            **(resume_handoff_context(carrier) if carrier else {}))
# Tool attempts can include:
metadata = action_metadata('order:42', attempt_id='try-2', retry_of_attempt_id='try-1')

The async client supports the same helpers with await client.handoff(...) and await client.begin(...).

Handoff records delegation before returning the carrier and the worker starts a separate trace linked by parent session. The carrier is correlation metadata, not a credential or verified business outcome. Use opaque identifiers, never customer content or secrets. Low-level createHandoffContext / create_handoff_context only serializes a carrier; it does not record delegation. V2 signal evidence treats detached outcomes as unresolved until their context is supported.

Account for worker completion

The v2 lifecycle adds a dispatch ID and explicit dispatch, acceptance, completion and parent-receipt records. Existing handoff() carriers remain supported. Use a new UUID for every worker execution; reuse it only for delivery retries of the same dispatch. Execution retries need a new dispatch ID and attempt_id, optionally linked with retry_of_attempt_id.

from uuid import uuid4
from sentrial import resume_handoff_context

# Parent: record intent before sending the carrier through your queue/RPC.
carrier = parent_client.dispatch_handoff(parent_session_id, "order:42", str(uuid4()))

# Worker: initialize its own session from the received carrier.
worker = worker_client.begin(
    user_id="example", event="order-worker", **resume_handoff_context(carrier)
)
if worker.session_id:
    worker_client.accept_handoff(worker.session_id, carrier)
    # Execute the work here. Report the actual execution outcome.
    worker_client.finish_handoff(worker.session_id, carrier, "completed")
worker.finish(output="Worker finished")

# Parent: only after receiving the terminal result through your application.
# Include the worker session ID and actual outcome in that result.
if worker.session_id:
    parent_client.receive_handoff(parent_session_id, carrier, worker.session_id, "completed")

AsyncSentrialClient exposes the same methods with await. Outcomes are completed, failed or cancelled; they describe execution, not verified business success. Emit completion before closing the worker and receipt before closing the parent. Never manufacture completion in an unconditional finally; missing records must remain unresolved. A returning telemetry call is not a collector or queue acknowledgment. Continue flushing/closing clients, inspecting delivery diagnostics, and running application work when optional telemetry session creation fails.

The internal reader checks all four steps, worker identity, conflicting outcomes and SDK producer sequence. It does not use cross-machine clocks to establish order. Combined signal evidence remains partial. See the handoff verification contract.

Report experiment exposure explicitly

Variant lookup and session creation link assignments (including holdouts), but do not claim the configuration was used. After applying the returned settings, call:

assignment = client.get_variant(
    "support-prompt", agent_name="Support agent", user_id="stable-user-id"
)
session_id = client.create_session(
    name="Support request", agent_name="Support agent", user_id="stable-user-id"
)
# Apply the returned settings in your application before recording exposure.
if session_id and assignment.get("assignmentId") and assignment.get("configurationHash"):
    receipt = client.record_experiment_exposure(
        assignment_id=assignment["assignmentId"],
        session_id=session_id,
        variant=assignment["variant"],
        configuration_hash=assignment["configurationHash"],
    )

The async client supports the same method with await. The configured module client also exposes sentrial.record_experiment_exposure(...). The SDK flushes pending telemetry before reporting. None means fail-silent delivery failure; strict mode raises errors. Retries with the same fields return the original receipt, including its server arrival time. This is application-reported use, not provider attestation or an inference call.

Report holdouts when their returned configuration was used, even though active is false. Historical assignments without a configuration digest cannot produce a matching receipt. Conflicting reports return 409; deleted session sources return 410 and remain as tombstones. Reports for already issued assignments can arrive after a run stops.

For overlapping sessions, use the explicit session_id argument to get_variant, or experiment_assignment_ids when creating a session. Automatic linking targets the next session created on that client; it does not infer concurrent request ownership. A session may have only one assignment identity for each experiment.

Assignment identity and telemetry acknowledgment

An experiment's declared assignment unit is required: supply userId/user_id for user assignment, convoId/convo_id for conversation assignment, or anonymousId/anonymous_id for anonymous assignment. Another identifier or a session ID will not substitute for it. Missing or whitespace-only identity returns active:false, reason:'missing_identity', the required assignmentUnit, and your default variant. No assignment is created. Preserve a stable identifier across requests. Pass that same user or conversation identifier when creating the session. A conflicting session identifier rejects the link; once linked, it cannot be changed or cleared. Assignments from different experiments must also agree on each shared assignment unit. An anonymous assignment has no corresponding session identity field, so its consistency is checked against the session's other anonymous assignments. This does not verify that the caller's identifier represents a real person or remains stable outside this session.

New assignments report identityProvenance:'declared' and their immutable assignmentUnit. Existing assignments report identityProvenance:'legacy_unknown' when their historical fallback identity cannot be established. A lookup does not upgrade that provenance. Start a new experiment for a prospectively defined identity/analysis plan.

Native telemetry commits a session's assignment links in the same transaction as its session/events. A failed link fails the request, and a retry preserves the original trace identity. Unknown or foreign assignments cannot be silently skipped. Explicit assignment lists must contain at most 64 UUIDs; malformed or oversized lists receive HTTP 400 before any trace in that batch writes. UUIDs are deduplicated without dropping distinct entries. A valid list whose assignment was deleted or is unavailable fails instead of returning a false success; correct the references before retrying a permanent error. The existing SDK fail-silent/strict error policy still applies.