Context Managers

AgentOps provides native context manager support for traces, allowing you to use Python’s with statement for automatic trace lifecycle management. This approach ensures traces are properly started and ended, even when exceptions occur.

Basic Usage

The simplest way to use context managers is with the start_trace() function:

import agentops

# Initialize AgentOps
agentops.init(api_key="your-api-key")

# Use context manager for automatic trace management
with agentops.start_trace("my_workflow") as trace:
    # Your code here
    print("Processing data...")
    # Trace automatically ends when exiting the with block

The trace will automatically:

Start when entering the with block
End with “Success” status when exiting normally
End with “Error” status if an exception occurs
Clean up resources properly in all cases

Advanced Usage

Traces with Tags

You can add tags to traces for better organization and filtering:

import agentops

agentops.init(api_key="your-api-key")

# Using list tags
with agentops.start_trace("data_processing", tags=["batch", "production"]):
    process_batch_data()

# Using dictionary tags for more structured metadata
with agentops.start_trace("user_request", tags={
    "user_id": "12345",
    "request_type": "query",
    "priority": "high"
}):
    handle_user_request()

Parallel Traces

Context managers create independent parallel traces, not parent-child relationships:

import agentops

agentops.init(api_key="your-api-key")

# Sequential parallel traces
with agentops.start_trace("task_1"):
    print("Task 1 executing")

with agentops.start_trace("task_2"):
    print("Task 2 executing")

# Nested context managers create parallel traces
with agentops.start_trace("outer_workflow"):
    print("Outer workflow started")
    
    with agentops.start_trace("inner_task"):
        print("Inner task executing (parallel to outer)")
    
    print("Outer workflow continuing")

Exception Handling

Context managers automatically handle exceptions and set appropriate trace states:

import agentops

agentops.init(api_key="your-api-key")

# Automatic error handling
try:
    with agentops.start_trace("risky_operation"):
        # This will automatically set trace status to "Error"
        raise ValueError("Something went wrong")
except ValueError as e:
    print(f"Caught error: {e}")
    # Trace has already been ended with Error status

# Graceful degradation pattern
try:
    with agentops.start_trace("primary_service"):
        result = call_primary_service()
except ServiceUnavailableError:
    with agentops.start_trace("fallback_service"):
        result = call_fallback_service()

Concurrent Execution

Context managers work seamlessly with threading and asyncio:

import agentops
import threading

agentops.init(api_key="your-api-key")

# With threading
def worker_function(worker_id):
    with agentops.start_trace(f"worker_{worker_id}"):
        # Each thread gets its own independent trace
        process_work(worker_id)

threads = []
for i in range(3):
    thread = threading.Thread(target=worker_function, args=(i,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

import agentops
import asyncio

agentops.init(api_key="your-api-key")

# With asyncio
async def async_task(task_id):
    with agentops.start_trace(f"async_task_{task_id}"):
        await asyncio.sleep(0.1)  # Simulate async work
        return f"result_{task_id}"

async def main():
    tasks = [async_task(i) for i in range(3)]
    results = await asyncio.gather(*tasks)
    return results

# Run async tasks
results = asyncio.run(main())

Production Patterns

API Endpoint Monitoring

import agentops
from flask import Flask, request

app = Flask(__name__)
agentops.init(api_key="your-api-key")

@app.route('/api/process', methods=['POST'])
def process_request():
    # Create trace for each API request
    with agentops.start_trace("api_request", tags={
        "endpoint": "/api/process",
        "method": "POST",
        "user_id": request.headers.get("user-id")
    }):
        try:
            data = request.get_json()
            result = process_data(data)
            return {"status": "success", "result": result}
        except Exception as e:
            # Exception automatically sets trace to Error status
            return {"status": "error", "message": str(e)}, 500

Batch Processing

import agentops

agentops.init(api_key="your-api-key")

def process_batch(items):
    with agentops.start_trace("batch_processing", tags={
        "batch_size": len(items),
        "batch_type": "data_processing"
    }):
        successful = 0
        failed = 0
        
        for item in items:
            try:
                with agentops.start_trace("item_processing", tags={
                    "item_id": item.get("id"),
                    "item_type": item.get("type")
                }):
                    process_item(item)
                    successful += 1
            except Exception as e:
                failed += 1
                print(f"Failed to process item {item.get('id')}: {e}")
        
        print(f"Batch completed: {successful} successful, {failed} failed")

Retry Logic

import agentops
import time

agentops.init(api_key="your-api-key")

def retry_operation(operation_name, max_retries=3):
    for attempt in range(max_retries):
        try:
            with agentops.start_trace(f"{operation_name}_attempt_{attempt + 1}", tags={
                "operation": operation_name,
                "attempt": attempt + 1,
                "max_retries": max_retries
            }):
                # Your operation here
                result = perform_operation()
                return result  # Success - exit retry loop
                
        except Exception as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Attempt {attempt + 1} failed: {e}. Retrying in {wait_time}s...")
                time.sleep(wait_time)
            else:
                print(f"All {max_retries} attempts failed")
                raise

Backward Compatibility

Context managers are fully backward compatible with existing AgentOps code patterns:

import agentops

agentops.init(api_key="your-api-key")

# Manual trace management (legacy)
trace = agentops.start_trace("manual_trace")
# ... your code ...
agentops.end_trace(trace, "Success")

import agentops

agentops.init(api_key="your-api-key")

# Context manager (new, recommended)
with agentops.start_trace("context_managed_trace") as trace:
    # ... your code ...
    pass  # Automatically ended

import agentops

agentops.init(api_key="your-api-key")

# Accessing trace properties
with agentops.start_trace("property_access") as trace:
    span = trace.span  # Access underlying span
    trace_id = trace.span.get_span_context().trace_id

import agentops

agentops.init(api_key="your-api-key")

# Mixed usage
trace = agentops.start_trace("mixed_usage")
try:
    with trace:  # Use existing trace as context manager
        # ... your code ...
        pass
except Exception:
    agentops.end_trace(trace, "Error")

Examples

For complete working examples, see the following files in the AgentOps repository:

Basic Usage

Simple context manager patterns and error handling

Parallel Traces

Sequential, nested, and concurrent trace patterns

Error Handling

Exception handling, retry patterns, and graceful degradation

Production Patterns

API endpoints, batch processing, microservices, and monitoring

These examples demonstrate real-world usage patterns and best practices for using AgentOps context managers in production applications.

API Reference

For detailed API information, see the SDK Reference documentation.

Getting Started

Concepts

Integrations

Usage

Self-Hosting

Other Info

Context Managers

Context Managers

Basic Usage

Advanced Usage

Traces with Tags

Parallel Traces

Exception Handling

Concurrent Execution

Production Patterns

API Endpoint Monitoring

Batch Processing

Retry Logic

Backward Compatibility

Examples

Basic Usage

Parallel Traces

Error Handling

Production Patterns

API Reference

​Context Managers

​Basic Usage

​Advanced Usage

​Traces with Tags

​Parallel Traces

​Exception Handling

​Concurrent Execution

​Production Patterns

​API Endpoint Monitoring

​Batch Processing

​Retry Logic

​Backward Compatibility

​Examples

Basic Usage

Parallel Traces

Error Handling

Production Patterns

​API Reference

Context Managers

Basic Usage

Advanced Usage

Traces with Tags

Parallel Traces

Exception Handling

Concurrent Execution

Production Patterns

API Endpoint Monitoring

Batch Processing

Retry Logic

Backward Compatibility

Examples

API Reference