Core Concepts

from agentbasis import trace

@trace
def process_user_input(text):
    # This entire function execution is recorded
    result = complex_logic(text)
    return result

from agentbasis import trace

@trace
async def fetch_and_process(url):
    data = await fetch_data(url)
    return process(data)

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

import agentbasis

current_user = agentbasis.get_user()
current_session = agentbasis.get_session()
current_conversation = agentbasis.get_conversation()
current_metadata = agentbasis.get_metadata()

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

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

import agentbasis

# Force send all pending telemetry
success = agentbasis.flush(timeout_millis=30000)

if success:
    print("All data sent")
else:
    print("Flush timed out")

import agentbasis

# Flush remaining data and shut down
agentbasis.shutdown()