The AgentBasis SDK provides powerful primitives for manually tracking your application logic and associating data with specific users.
Manual Tracing
Use the @trace decorator to automatically track any function execution as a span. This captures arguments, return values, and execution time.
from agentbasis import trace
@trace
def process_user_input(text):
# This entire function execution is recorded
result = complex_logic(text)
return result
Async Functions
The @trace decorator works with both sync and async functions:
from agentbasis import trace
@trace
async def fetch_and_process(url):
data = await fetch_data(url)
return process(data)
User & Session Context
Context allows you to associate traces with specific users or sessions, enabling you to debug issues per-user or view session-based analytics.
Global Context
Set context globally, which applies to all subsequent operations until changed:
import agentbasis
# Set the current user (e.g., from your auth system)
agentbasis.set_user("user-123")
# Optionally set session and conversation IDs
agentbasis.set_session("session-abc")
agentbasis.set_conversation("conv-123")
# Set custom metadata (e.g., subscription plan, feature flags)
agentbasis.set_metadata({"plan": "pro", "feature_flag": "new_ui"})
# All subsequent calls will be tagged with this context
Reading Context
You can read the current context values:
import agentbasis
current_user = agentbasis.get_user()
current_session = agentbasis.get_session()
current_conversation = agentbasis.get_conversation()
current_metadata = agentbasis.get_metadata()
Scoped Context (Context Manager)
For multi-threaded or async applications, use the context manager to scope data to a specific block:
from agentbasis import context
with context(user_id="user-123", session_id="session-abc"):
# All traces in this block include the context
response = client.chat.completions.create(...)
# Context is automatically cleared after the block
Scoped Context (Decorator)
Use @with_context to set context for an entire function:
from agentbasis import with_context
@with_context(user_id="user-123", session_id="session-abc")
def handle_request():
# All traces in this function include the context
response = client.chat.completions.create(...)
return response
Flush & Shutdown
Flushing Data
The SDK batches and sends data asynchronously. If you need to ensure all data is sent at a specific point, use flush():
import agentbasis
# Force send all pending telemetry
success = agentbasis.flush(timeout_millis=30000)
if success:
print("All data sent")
else:
print("Flush timed out")
Common use cases for flush():
- Before a critical operation
- At specific checkpoints in your application
- In serverless functions before the function terminates
Shutdown
The SDK automatically shuts down and flushes on normal Python exit. If you need to manually shut down:
import agentbasis
# Flush remaining data and shut down
agentbasis.shutdown()
shutdown() is idempotent - calling it multiple times is safe.
How It Works
- OpenTelemetry: We use OTel under the hood for maximum compatibility with observability tools.
- Spans: Every action (function call, LLM request) is recorded as a Span with timing, inputs, and outputs.
- Transport: Data is batched and sent asynchronously, ensuring minimal impact on your application’s latency.
- Context Propagation: User, session, and metadata are automatically attached to all spans.