Skip to main content
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.