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 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 )
sentrial.begin()Begin tracking an interaction. Returns an Interaction object.
Custom event ID (auto-generated if omitted).
Conversation ID for grouping.
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 = 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" )
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 )
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.
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()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 ) -> dict
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 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)
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 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)
Event Batching Queue tracking calls and flush periodically to reduce HTTP overhead.
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()
PII Redaction Automatically redact sensitive data before it leaves your infrastructure.
# 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 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 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()Flush any queued events immediately. No-op if batching is not enabled.
client.shutdown()Shut down the event batcher, flushing remaining events. Call this before your process exits to ensure all events are sent.
LangChain Integration 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.
Sessions API REST API reference.
