Skip to main content
The Muxx Python SDK provides decorators for clean, declarative observability.

​
@observe

The @observe decorator is the simplest way to add tracing: 
from muxx import observe
from openai import OpenAI

client = OpenAI()

@observe()
def get_completion(prompt: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

​
With Custom Name

@observe("summarize-text")
def summarize(text: str) -> str:
    # Function name in dashboard will be "summarize-text"
    pass

​
With Metadata

@observe("user-chat", metadata={"feature": "support"})
def handle_chat(user_id: str, message: str):
    pass

​
@trace

Creates a new trace for the function: 
from muxx import trace

@trace("document-pipeline")
def process_document(doc_id: str):
    # All operations inside create a new trace
    extract_text(doc_id)
    analyze_content(doc_id)
    generate_summary(doc_id)

​
Trace Options

@trace(
    name="user-request",
    metadata={"environment": "production"},
    capture_input=True,   # Log function arguments
    capture_output=True   # Log return value
)
def handle_request(request_data: dict):
    pass

​
@span

Creates a span within the current trace: 
from muxx import trace, span

@trace("multi-step-process")
def process():

    @span("step-1")
    def step_one():
        pass

    @span("step-2")
    def step_two():
        pass

    step_one()
    step_two()

​
@generation

Explicitly marks a function as an LLM generation: 
from muxx import generation

@generation(model="gpt-4o", provider="openai")
def call_llm(prompt: str):
    # Useful when using a custom LLM client
    # that Muxx doesn't automatically detect
    pass

​
Combining Decorators

Decorators can be combined: 
from muxx import trace, span, observe

@trace("api-request")
def handle_api_request(request):

    @span("validate")
    def validate():
        pass

    @span("process")
    @observe()  # This span is also observed
    def process():
        # LLM call here
        pass

    validate()
    return process()

​
Async Decorators

All decorators work with async functions: 
from muxx import trace, span

@trace("async-pipeline")
async def async_pipeline():

    @span("async-step")
    async def async_step():
        await some_async_operation()

    await async_step()

​
Class Methods

Decorators work on class methods: 
from muxx import trace, observe

class DocumentProcessor:

    @trace("process-document")
    def process(self, document: str):
        return self._summarize(document)

    @observe("summarize")
    def _summarize(self, text: str):
        # LLM call
        pass

​
Error Handling

Decorators automatically capture errors: 
from muxx import observe

@observe("risky-operation")
def risky_operation():
    raise ValueError("Something went wrong")
    # Error is captured in the trace with full stack trace
You can also add custom error handling: 
from muxx import observe, get_current_span

@observe("operation")
def operation():
    try:
        # risky code
        pass
    except Exception as e:
        span = get_current_span()
        span.set_error(e, metadata={"recovery_attempted": True})
        raise