Python SDK

pip install sentrial

# With LangChain support
pip install sentrial langchain-core

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)

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
)

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.finish(
    output: str = None,              # Output/response from interaction
    success: bool = True,            # Whether it succeeded
    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 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.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(
    reasoning="Will search KB before escalating",
    alternatives=["escalate_to_human", "ask_clarifying_question"],
    confidence=0.92
)

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)
)

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

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
    trace_id: str = None,            # Trace ID for distributed tracing
    span_id: str = None,             # Span ID for distributed tracing
    metadata: dict = None            # Additional metadata
) -> dict

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(
    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
) -> dict

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

from openai import OpenAI
from sentrial import wrap_openai, configure

configure(api_key="sentrial_live_xxx")

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

interaction = sentrial.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)

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

from sentrial import SentrialClient

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 async_client.begin(user_id="user_123", event="chat") as interaction:
    response = await agent.run("Hello")
    interaction.set_output(response)

from sentrial import configure, BatcherConfig

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

# Advanced: customize flush behavior
configure(
    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()

# 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=["user_input", "assistant_output", "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 = SentrialClient.calculate_openai_cost(
    model="gpt-4o",
    input_tokens=1000,
    output_tokens=500
)
# Returns: 0.0075 (USD)

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

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

import sentrial

# Async module-level API
sentrial.configure(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.flush()

client.shutdown()