Skip to main content
While Polos automatically traces LLM calls, tool invocations, and workflow steps, you can add custom spans to capture additional context specific to your application.

Creating a custom span

Use ctx.step.trace() to create a custom span:
from polos import workflow, WorkflowContext

@workflow
async def my_workflow(ctx: WorkflowContext, input: MyInput):
    # Custom span for a database query
    with ctx.step.trace("database_query", {"table": "users"}) as span:
        # Your code here
        result = await ctx.step.run("query_database", query_database, "SELECT * FROM users")

        # Add attributes to the span
        span.set_attribute("row_count", len(result))

    return result
The first argument is the span name, and the second (optional) argument is a dictionary of initial attributes.

Setting attributes

You can add metadata to spans using attributes:
with ctx.step.trace("external_api_call") as span:
    response = await ctx.step.run("external_api", call_external_api, url)

    # Set a single attribute
    span.set_attribute("status_code", response.status_code)
    span.set_attribute("success", response.ok)

    # Set multiple attributes at once
    span.set_attributes({
        "url": url,
        "response_size": len(response.content),
        "content_type": response.headers.get("content-type"),
    })

Adding events

Events mark specific moments within a span:
with ctx.step.trace("data_processing") as span:
    span.add_event("started_validation")

    await ctx.step.run("validate_data", validate_data, data)

    span.add_event("validation_complete", attributes={
        "records_validated": len(data),
        "errors_found": 0,
    })

    process_data(data)

    span.add_event("processing_complete")

Nested spans

Spans can be nested to show hierarchical relationships:
@workflow
async def my_workflow(ctx: WorkflowContext, input: MyInput):
    with ctx.step.trace("parent_operation") as parent_span:
        # Do some work
        await ctx.step.run("step_one", step_one_func)

        # Nested span within the parent
        with ctx.step.trace("child_operation") as child_span:
            child_span.set_attribute("nested", True)
            await ctx.step.run("do_child_work", do_child_work)

        await ctx.step.run("step_two", step_two_func)
You can also create spans within step functions:
async def process_data(ctx: WorkflowContext, data: dict):
    with ctx.step.trace("data_transformation", {"input_size": len(data)}) as span:
        transformed = transform(data)

        span.set_attributes({
            "output_size": len(transformed),
            "success": True,
        })
        span.add_event("transformation_complete")

        return transformed
These custom spans appear alongside automatic traces in the Polos dashboard, giving you complete visibility into your workflow execution.