@workflow | Respan Docs

Overview

The @workflow decorator creates a root trace span. All nested @task, @agent, and @tool spans are captured as children under this workflow.

from respan import workflow

Parameters

Parameter	Type	Default	Description
`name`	`str \| None`	Function name	Display name for the workflow span.
`version`	`int \| None`	`None`	Version number for the workflow.
`method_name`	`str \| None`	`None`	Required when decorating a class. Specifies which method to use as the entry point.
`processors`	`str \| List[str] \| None`	`None`	Route this span to specific named processors. See add_processor.
`export_filter`	`FilterParamDict \| None`	`None`	Filter dict to control which spans are exported. Uses AND logic. See Export filtering.

Function usage

from respan import Respan, workflow, task
respan = Respan(api_key="your-api-key")
@task(name="extract")
def extract():
    return {"records": [1, 2, 3]}
@task(name="transform")
def transform(data):
    return [x * 2 for x in data["records"]]
@workflow(name="data_pipeline")
def data_pipeline():
    data = extract()
    return transform(data)
print(data_pipeline())  # [2, 4, 6]

Class usage

When decorating a class, use method_name to specify the entry point. Calling that method creates the workflow span.

from respan import Respan, workflow, task
from openai import OpenAI
respan = Respan(api_key="your-api-key")
client = OpenAI()
@workflow(name="analysis_workflow", method_name="run")
class Analyzer:
    @task(name="analyze")
    def analyze(self, nums):
        return sum(nums)
    def run(self):
        return self.analyze([1, 2, 3])
print(Analyzer().run())  # 6

Processor routing

Use processors to send workflow spans to specific exporters only.

@workflow(name="debug_workflow", processors="debug")
def debug_workflow():
    return "only exported to the 'debug' processor"
@workflow(name="multi_export", processors=["debug", "analytics"])
def multi_export():
    return "exported to both 'debug' and 'analytics' processors"

Export filtering

Use export_filter to conditionally export spans based on their attributes. All conditions use AND logic.

from respan import workflow
# Only export workflow spans that ended with an error
@workflow(
    name="monitored_workflow",
    export_filter={"status_code": {"operator": "", "value": "ERROR"}}
)
def monitored_workflow():
    return "only exported if the span has ERROR status"

Supported operators: "" (equals), "not", "gt", "gte", "lt", "lte", "contains", "icontains", "startswith", "endswith", "regex", "in", "not_in", "empty", "not_empty".

Best practices

Use descriptive workflow names for easy navigation in the Traces dashboard
Keep workflows coarse-grained — use @task for internal steps
One workflow per user request or pipeline run