Python SDK

Learn how to instrument your Python AI agents for deep debugging and decision visibility.

Installation

pip install opswald

Quick Start

import opswald

# Initialize with your API key
opswald.init(api_key='your-key', base_url='https://api.opswald.com')

# Create a trace around your agent run
with opswald.trace('my-agent-run') as t:
    with opswald.span('reasoning', kind='llm_call', provider='openai', model='gpt-4o') as s:
        s.set_input({'prompt': 'Analyze this data...'})

        # Your AI logic here
        result = openai.chat.completions.create(...)

        s.set_output({'analysis': result.choices[0].message.content})
        s.set_tokens(input_tokens=120, output_tokens=340)

Auto-Instrumentation

The fastest way to get started is auto-instrumentation:

import opswald
import openai
import anthropic

# Initialize
opswald.init(api_key='your-key')

# Create client instances and instrument them
openai_client = openai.OpenAI()
anthropic_client = anthropic.Anthropic()

opswald.instrument(openai_client)
opswald.instrument(anthropic_client)

# Now all calls on these clients are automatically traced
response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)
# ^ This call is automatically captured as a span

Manual Traces and Spans

For full control over what gets captured:

Traces

A trace represents a complete agent run or workflow:

# Context manager (recommended)
with opswald.trace('user-query-pipeline') as trace:
    # Your agent logic here
    pass

# Function decorator
@opswald.trace_function
def process_user_query(query):
    # This entire function becomes a trace
    return analysis

Spans

Spans represent individual steps within a trace:

with opswald.trace('data-analysis') as t:
    # LLM reasoning step
    with opswald.span('extract-insights', kind='llm_call') as s:
        s.set_input({'raw_data': data})
        insights = analyze_data(data)
        s.set_output({'insights': insights})

    # Tool usage step
    with opswald.span('search-database', kind='tool_call') as s:
        s.set_input({'query': insights.search_terms})
        results = database.search(insights.search_terms)
        s.set_output({'results': results})

    # Error handling
    try:
        risky_operation()
    except Exception as e:
        with opswald.span('error-recovery', kind='error') as s:
            s.set_error({'error': str(e)})
            fallback_result = handle_error(e)
            s.set_output({'fallback': fallback_result})

Span Types

Use the appropriate kind for different operations:

llm_call — LLM API calls (OpenAI, Anthropic, etc.)
tool_call — Function calls, API calls, database queries
error — Error handling and recovery
custom — Everything else

Metadata and Context

Add rich context to your spans:

with opswald.span('user-intent-analysis',
                  kind='llm_call',
                  provider='openai',
                  model='gpt-4o',
                  temperature=0.1) as s:

    # Input/output
    s.set_input({'user_message': message, 'chat_history': history})
    s.set_output({'intent': intent, 'confidence': 0.92})

    # Token usage
    s.set_tokens(input_tokens=150, output_tokens=45)

    # Custom metadata
    s.set_metadata({
        'user_id': user.id,
        'session_id': session.id,
        'model_version': 'v2.1'
    })

Real-World Example

Here’s how to trace a complete RAG (Retrieval-Augmented Generation) pipeline:

import opswald
import openai
from vector_db import VectorDB

opswald.init(api_key='your-key')
db = VectorDB()
client = openai.OpenAI()

def rag_pipeline(user_question):
    with opswald.trace('rag-query') as t:
        # 1. Generate embeddings for user question
        with opswald.span('embed-question', kind='llm_call', model='text-embedding-3-small') as s:
            s.set_input({'question': user_question})
            embedding = client.embeddings.create(
                model="text-embedding-3-small",
                input=user_question
            ).data[0].embedding
            s.set_output({'embedding_length': len(embedding)})

        # 2. Search vector database
        with opswald.span('vector-search', kind='tool_call') as s:
            s.set_input({'embedding': embedding[:5]})  # First 5 dims for brevity
            docs = db.similarity_search(embedding, top_k=5)
            s.set_output({'num_docs': len(docs), 'relevance_scores': [d.score for d in docs]})

        # 3. Generate final answer
        with opswald.span('generate-answer', kind='llm_call', model='gpt-4o') as s:
            context = "\n".join([doc.text for doc in docs])
            prompt = f"Context:\n{context}\n\nQuestion: {user_question}\n\nAnswer:"

            s.set_input({'context_length': len(context), 'question': user_question})

            response = client.chat.completions.create(
                model="gpt-4o",
                messages=[{"role": "user", "content": prompt}]
            )

            answer = response.choices[0].message.content
            s.set_output({'answer': answer})
            s.set_tokens(
                input_tokens=response.usage.prompt_tokens,
                output_tokens=response.usage.completion_tokens
            )

            return answer

# Usage
answer = rag_pipeline("What are the benefits of vector databases?")

Best Practices

1. Meaningful Span Names

Use descriptive names that explain what the step does:

# ❌ Generic
with opswald.span('llm'):
    ...

# ✅ Descriptive
with opswald.span('extract-key-entities', kind='llm_call'):
    ...

2. Capture Decisions

Use metadata to capture the reasoning behind decisions:

with opswald.span('route-user-query', kind='custom') as s:
    confidence = classify_intent(query)
    route = 'expert' if confidence > 0.8 else 'general'

    s.set_metadata({
        'classification_confidence': confidence,
        'routing_decision': route,
        'reasoning': f'Confidence {confidence:.2f} {">" if confidence > 0.8 else "<="} 0.8 threshold'
    })

3. Error Recovery Patterns

Always trace error handling so you can debug failures:

try:
    result = risky_ai_operation()
except Exception as e:
    with opswald.span('error-recovery', kind='error') as s:
        s.set_error({'error': str(e), 'error_type': type(e).__name__})

        # Capture recovery logic
        fallback = get_fallback_response()
        s.set_output({'fallback_used': True, 'fallback': fallback})

        result = fallback

4. Nested Workflows

Use nested spans to capture sub-workflows:

with opswald.trace('content-generation') as t:
    with opswald.span('research-phase', kind='custom') as research:
        with opswald.span('web-search', kind='tool_call') as s:
            # Search implementation
            pass
        with opswald.span('summarize-results', kind='llm_call') as s:
            # Summarization implementation
            pass

    with opswald.span('writing-phase', kind='custom') as writing:
        with opswald.span('generate-outline', kind='llm_call') as s:
            # Outline generation
            pass
        with opswald.span('write-content', kind='llm_call') as s:
            # Content writing
            pass

Configuration

Customize the SDK behavior:

opswald.init(
    api_key='your-key',
    base_url='https://api.opswald.com',  # Custom endpoint
    batch_size=50,                       # Spans per batch
    flush_interval_ms=5000,             # Auto-flush interval
    max_retries=3,                      # Retry failed requests
    debug=True                          # Enable debug logging
)

Graceful Shutdown

Always flush pending spans before your application exits:

import opswald
import atexit

# Automatic cleanup on exit
atexit.register(opswald.shutdown)

# Or manual cleanup
try:
    # Your application code
    run_agent()
finally:
    opswald.shutdown()  # Ensures all spans are sent