Python SDK - Sentrial

Installation

pip install sentrial

# With LangChain support
pip install sentrial langchain-core

Quick Start

import sentrial

# Configure (or set SENTRIAL_API_KEY env var)
sentrial.configure(api_key="sentrial_live_xxx")

# Track an interaction
interaction = sentrial.begin(
    user_id="user_123",
    event="chat_message",
    input="User's question"
)

# Your agent logic...
response = agent.run(user_input)

# Finish with outcome
interaction.finish(output=response, success=True)

Module Functions

`sentrial.configure()`

Configure the default Sentrial client.

sentrial.configure(
    api_key: str = None,         # API key (or use SENTRIAL_API_KEY env)
    api_url: str = None,         # API URL (default: https://api.sentrial.com)
    fail_silently: bool = True,  # If True, SDK errors are logged but won't crash your app
    pii: bool | PiiConfig = None,  # PII redaction (pass True to auto-fetch from server)
    batching: bool | BatcherConfig = None,  # Event batching for high-throughput agents
)

Environment Variables & Transport

SENTRIAL_API_KEY=sentrial_live_xxxxxxxxxxxxx
SENTRIAL_API_URL=https://api.sentrial.com  # Optional

The SDK ingests over OTLP (/api/otel/v1/traces) — this is the sole ingestion path. The OTLP path 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.

`sentrial.begin()`

Begin tracking an interaction. Returns an Interaction object.

string

required

External user ID.

string

required

Event type / agent name.

string

Input data.

string

Custom event ID (auto-generated if omitted).

string

Conversation ID for grouping.

dict

Additional metadata.

interaction = sentrial.begin(
    user_id="user_12345",
    event="support_request",
    input="Help me reset my password",
    convo_id="conv_789",
    metadata={"channel": "web_chat"}
)

Interaction Class

Returned by sentrial.begin(). Provides methods to track events and finish the interaction.

`interaction.finish()`

Finish the interaction and record final metrics.

interaction.finish(
    output: str = None,              # Output/response from interaction
    success: bool = None,            # See below — defaults to tracked state
    failure_reason: str = None,      # Reason if success=False
    estimated_cost: float = None,    # Cost in USD
    custom_metrics: dict = None,     # Custom metrics
    duration_ms: int = None,         # Duration in milliseconds (auto-calculated if omitted)
    prompt_tokens: int = None,       # Prompt tokens used
    completion_tokens: int = None,   # Completion tokens used
    total_tokens: int = None         # Total tokens used
)

success defaults to None, which means “use the tracked state”: True normally, but if interaction.track_error() was called, the default flips to False (with failure_reason defaulting to the tracked error message). Pass success=True/False explicitly to override.

# Success example
interaction.finish(
    output="Password reset email sent!",
    success=True,
    estimated_cost=0.023,
    custom_metrics={"satisfaction": 4.5}
)

# Failure example
interaction.finish(
    success=False,
    failure_reason="User not found in database"
)

`interaction.set_metadata()` / `interaction.set_model()`

Update the session after begin() — both are merged into the session on finish() (the finalize re-post). set_metadata merges keys into the session metadata (incoming values win); set_model records the session-level model (an explicit finish()-time model still wins).

interaction.set_metadata({"experiment": "v2", "region": "us-east"})
interaction.set_model("claude-sonnet-4-5")

Also available client-level as client.set_session_metadata(session_id, metadata) and client.set_session_model(session_id, model). begin() additionally accepts name=... (session title override; auto-named {event}:{id} otherwise) and model=....

`interaction.track_tool_call()`

Track a tool call within the interaction.

interaction.track_tool_call(
    tool_name="search_knowledge_base",
    tool_input={"query": "password reset"},
    tool_output={"results": ["KB-001", "KB-002"]},
    reasoning="User needs password help",
    estimated_cost=0.001
)

`interaction.track_decision()`

Track an agent decision.

interaction.track_decision(
    reasoning="Will search KB before escalating",
    alternatives=["escalate_to_human", "ask_clarifying_question"],
    confidence=0.92
)

`interaction.report_diagnostic(report)`

Record an agent’s observation about missing context, a tool problem, a missing capability, or uncertain completion. Call this explicitly while the interaction is active. It does not call a model, change session success, or create a confirmed signal or training label.

Agent reports are included in the unreleased SDK changes. Staging acceptance is complete; package publication is pending. Use matching SDK, collector and trace UI builds to test this feature.

interaction.report_diagnostic({
    "reportId": "claim-context-1",
    "category": "missing_context",
    "state": "reported",
    "description": "The policy number needed for this claim is missing.",
})

# If the agent later observes recovery, retain a second observation with the same report ID.
interaction.report_diagnostic({
    "reportId": "claim-context-1",
    "category": "missing_context",
    "state": "recovered",
    "description": "The policy number was supplied; claim submission has not been verified.",
})

await interaction.report_diagnostic({
    "reportId": "claim-context-1",
    "category": "missing_context",
    "state": "reported",
    "description": "The policy number needed for this claim is missing.",
})

await interaction.report_diagnostic({
    "reportId": "claim-context-1",
    "category": "missing_context",
    "state": "recovered",
    "description": "The policy number was supplied; claim submission has not been verified.",
})

The client equivalents are client.report_diagnostic(session_id, report) and await async_client.report_diagnostic(session_id, report). Use the same client that created the active session. They return None, which is not a durable-delivery receipt. Check transport diagnostics and the retained trace. Finished or degraded interactions do not accept new reports. The dictionary uses the same camelCase field names as the TypeScript SDK, including reportId and relatedSpanIds:

Field	Accepted values
`reportId`	Session-scoped opaque ID, 1–128 ASCII letters, digits, `.`, `_`, `:`, or `-`; starts with a letter or digit. Reuse it for subsequent observations of the same problem.
`category`	`missing_context`, `tool_problem`, `missing_capability`, or `uncertain_completion`.
`state`	`reported` or `recovered`. Both are assertions by the agent.
`description`	Nonblank text, at most 4,096 UTF-8 bytes.
`relatedSpanIds`	Optional list of up to eight distinct, nonzero, lowercase 16-digit hexadecimal OTel span IDs from this trace. References remain unverified.

The SDK sets the report revision and origin; do not supply extra fields. Configured metadata PII redaction, batching, retry and fail_silently behavior apply. Each call creates a separate event; recovery does not overwrite the original report or prove business success. The trace UI labels these Agent report and Reported recovered. Older collectors retain generic metadata without the validated report presentation. Absence of a report does not establish success.

SentrialClient Class

For full control over session management.

Constructor

from sentrial import SentrialClient

client = SentrialClient(
    api_key: str = None,   # API key (or SENTRIAL_API_KEY env)
    api_url: str = None    # API URL (or SENTRIAL_API_URL env)
)

`client.create_session()`

Create a new tracking session. Returns the session ID, or None if creation failed and the client is configured with fail_silently (the default).

session_id = client.create_session(
    name: str,                        # Descriptive name
    agent_name: str,                  # Agent identifier (for grouping)
    user_id: str,                     # External user ID
    parent_session_id: str = None,    # Parent session ID (for sub-sessions)
    convo_id: str = None,             # Conversation ID (for grouping)
    metadata: dict = None             # Custom metadata
) -> str | None

`client.track_tool_call()`

client.track_tool_call(
    session_id: str,
    tool_name: str,
    tool_input: dict,
    tool_output: dict,
    reasoning: str = None,
    estimated_cost: float = 0.0,
    tool_error: dict = None,         # Error details if the tool call failed
    token_count: int = None,         # Tokens used by the tool call
    model: str = None,               # LLM model name (for LLM-based tools)
    prompt_tokens: int = None,       # Prompt/input tokens
    completion_tokens: int = None,   # Completion/output tokens
    finish_reason: str = None,       # LLM finish reason
    trace_id: str = None,            # Deprecated — accepted for backward compatibility, no longer used by the OTLP transport
    span_id: str = None,             # Deprecated — accepted for backward compatibility, no longer used by the OTLP transport
    metadata: dict = None            # Additional metadata
) -> dict

`client.track_decision()`

client.track_decision(
    session_id: str,
    reasoning: str,
    alternatives: list[str] = None,
    confidence: float = None,
    estimated_cost: float = 0.0,
    token_count: int = None
) -> dict

`client.complete_session()`

Complete a session with final metrics.

client.complete_session(
    session_id: str,
    success: bool = True,
    failure_reason: str = None,
    estimated_cost: float = None,
    custom_metrics: dict = None,
    duration_ms: int = None,
    prompt_tokens: int = None,
    completion_tokens: int = None,
    total_tokens: int = None,
    user_input: str = None,            # End-user input for this session
    assistant_output: str = None       # Agent/assistant response
) -> None

`client.track_error()`

client.track_error(
    session_id: str,
    error_message: str,
    error_type: str = None,         # e.g. "ValueError", "APIError"
    tool_name: str = None,          # Tool that caused the error
    stack_trace: str = None,        # Stack trace for debugging
    metadata: dict = None
) -> dict

LLM Auto-Wrappers

Automatically track all LLM calls with token counts, cost, and latency.

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

configure(api_key="sentrial_live_xxx")

# Wrap your LLM client — all calls are now auto-tracked
client = wrap_openai(OpenAI())

interaction = begin(user_id="user_123", event="chat", input=user_query)

# This call is automatically tracked
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": user_query}]
)

interaction.finish(output=response.choices[0].message.content)

Also available: wrap_anthropic(client), wrap_google(model), wrap_llm(client) (auto-detect).

Decorators

from sentrial import tool, session, configure

configure(api_key="sentrial_live_xxx")

@tool("search_kb")
def search_knowledge_base(query: str) -> dict:
    """Automatically tracked — inputs, outputs, errors, duration."""
    return kb.search(query)

@session("support-agent")
def handle_request(user_id: str, message: str) -> str:
    """Auto-creates session, tracks all @tool calls inside."""
    results = search_knowledge_base(message)
    return format_response(results)

# Works with async too
@session("async-agent")
async def handle_async(user_id: str, message: str) -> str:
    results = await async_search(message)
    return results

Context Manager

from sentrial import SentrialClient, AsyncSentrialClient

client = SentrialClient(api_key="sentrial_live_xxx")

# Auto-finishes on exit, tracks success/failure
with client.begin(user_id="user_123", event="chat", input="Hello") as interaction:
    response = agent.run("Hello")
    interaction.set_output(response)

# Also works with async (use AsyncSentrialClient or module-level begin_async)
async_client = AsyncSentrialClient(api_key="sentrial_live_xxx")
async with await async_client.begin(user_id="user_123", event="chat") as interaction:
    response = await agent.run("Hello")
    interaction.set_output(response)

Event Batching

Queue tracking calls and flush periodically to reduce HTTP overhead.

from sentrial import SentrialClient, BatcherConfig

# Simple: just enable batching
client = SentrialClient(api_key="sentrial_live_xxx", batching=True)

# Advanced: customize flush behavior
client = SentrialClient(
    api_key="sentrial_live_xxx",
    batching=BatcherConfig(
        enabled=True,
        flush_interval=2.0,    # Flush every 2s (default: 1.0)
        flush_threshold=20,    # Flush after 20 events (default: 10)
        max_queue_size=500,    # Drop events if queue exceeds this (default: 1000)
    )
)

# Flush manually before shutdown
client.flush()
client.shutdown()

The module-level API supports the same options: sentrial.configure(api_key="...", batching=True).

PII Redaction

Automatically redact sensitive data before it leaves your infrastructure.

Auto-fetching (pii=True) requires the API key to have the pii-config:read scope. If the key lacks it the fetch is refused and the SDK falls back to the default redaction config (label mode, default fields, builtin patterns) — your org’s configured mode/fields/custom patterns are NOT applied. The SDK logs a one-time warning when this happens.

# Auto-fetch your org's PII config from the server
configure(api_key="sentrial_live_xxx", pii=True)

# Or configure locally
from sentrial.redact import PiiConfig, PiiBuiltinPatterns, PiiCustomPattern
from sentrial import SentrialClient

client = SentrialClient(
    api_key="sentrial_live_xxx",
    pii=PiiConfig(
        enabled=True,
        mode="label",            # "label" | "hash" | "remove"
        fields=["userInput", "assistantOutput", "metadata", "reasoning"],
        builtin_patterns=PiiBuiltinPatterns(
            emails=True,
            phones=True,
            ssns=True,
            credit_cards=True,
        ),
        custom_patterns=[
            PiiCustomPattern(label="api_key", pattern=r"sk-[a-zA-Z0-9]{32,}"),
        ],
    ),
)

Cost Calculation Helpers

Static methods to calculate LLM API costs.

`calculate_openai_cost()`

cost = SentrialClient.calculate_openai_cost(
    model="gpt-4o",
    input_tokens=1000,
    output_tokens=500
)
# Returns: 0.0075 (USD)

`calculate_anthropic_cost()`

cost = SentrialClient.calculate_anthropic_cost(
    model="claude-sonnet-4",
    input_tokens=1000,
    output_tokens=500
)
# Returns: 0.0105 (USD)

`calculate_google_cost()`

cost = SentrialClient.calculate_google_cost(
    model="gemini-2.0-flash",
    input_tokens=1000,
    output_tokens=500
)

Async Support

All SDK classes have async equivalents for use with asyncio.

import sentrial

# Async module-level API — use configure_async, not configure.
# (configure only sets up the sync client; begin_async would fall back to an
# env-var-only client, so an api_key passed to configure would never reach it.)
await sentrial.configure_async(api_key="sentrial_live_xxx")

interaction = await sentrial.begin_async(
    user_id="user_123",
    event="chat_message",
    input="User's question"
)

response = await agent.run(user_input)
await interaction.finish(output=response)

# Async client
from sentrial import AsyncSentrialClient

client = AsyncSentrialClient(api_key="sentrial_live_xxx")

session_id = await client.create_session(
    name="Async session",
    agent_name="my-agent",
    user_id="user_123"
)

await client.track_tool_call(
    session_id=session_id,
    tool_name="search",
    tool_input={"query": "test"},
    tool_output={"results": []}
)

await client.complete_session(session_id=session_id, success=True)

`client.close()`

The async client has no flush() / shutdown() — its only lifecycle method is close(). Call it before your process exits: it flushes any batched events, then closes the underlying HTTP client (and the Langfuse integration, if enabled).

await client.close()

LangChain Integration

Use SentrialCallbackHandler to automatically track all LLM calls, tool usage, and agent decisions in LangChain. Full LangChain Integration Guide →

Next Steps

LangChain Integration

Automatic tracking for LangChain agents.

Claude Code Integration

Auto-track Claude Agent SDK sessions and tool calls.

Mastra Integration

One-line setup for Mastra AI agents.

OpenAI Agents SDK

Mirror every Agents SDK run into a session — LLM calls, tools, handoffs, guardrails.

Sessions API

REST API reference.

​Installation

​Quick Start

​Module Functions

​sentrial.configure()

​Environment Variables & Transport

​sentrial.begin()

​Interaction Class

​interaction.finish()

​interaction.set_metadata() / interaction.set_model()

​interaction.track_tool_call()

​interaction.track_decision()

​interaction.report_diagnostic(report)

​SentrialClient Class

​Constructor

​client.create_session()

​client.track_tool_call()

​client.track_decision()

​client.complete_session()

​client.track_error()

​LLM Auto-Wrappers

​Decorators

​Context Manager

​Event Batching

​PII Redaction

​Cost Calculation Helpers

​calculate_openai_cost()

​calculate_anthropic_cost()

​calculate_google_cost()

​Async Support

​client.close()

​LangChain Integration

​Next Steps

LangChain Integration

Claude Code Integration

Mastra Integration

OpenAI Agents SDK

Sessions API

Installation

Quick Start

Module Functions

`sentrial.configure()`

Environment Variables & Transport

`sentrial.begin()`

Interaction Class

`interaction.finish()`

`interaction.set_metadata()` / `interaction.set_model()`

`interaction.track_tool_call()`

`interaction.track_decision()`

`interaction.report_diagnostic(report)`

SentrialClient Class

Constructor

`client.create_session()`

`client.track_tool_call()`

`client.track_decision()`

`client.complete_session()`

`client.track_error()`

LLM Auto-Wrappers

Decorators

Context Manager

Event Batching

PII Redaction

Cost Calculation Helpers

`calculate_openai_cost()`

`calculate_anthropic_cost()`

`calculate_google_cost()`

Async Support

`client.close()`

LangChain Integration

Next Steps