The @step decorator

from loomflow import step

@step marks an async def as a named step, so that when it runs inside a Workflow it gets its own loom.workflow.step span (when the workflow has telemetry=) and its own audit entries (when it has audit_log=). Use it to see the helper functions a workflow node calls as separate, timed steps in your trace and audit log. Control flow stays as plain if/await/return; outside a workflow the decorator does nothing.

Signature

def step(
    fn: Callable[..., Awaitable[Any]] | None = None,
    *,
    name: str | None = None,
) -> Any: ...

Use as either @step (uses the function’s __name__) or @step(name="custom-step-name").

What it does

When called during a Workflow.run / Workflow.stream, the decorated function:

1. Reads the workflow’s telemetry and audit log from the live run context.
2. Opens a loom.workflow.step span tagged with step=<name>, user_id, session_id, pattern="workflow" (when the workflow has telemetry=).
3. Writes a step_started audit entry, runs the wrapped function, then writes step_completed, or step_failed with the error if it raised (when the workflow has audit_log=). Each entry’s payload names the workflow and the step ({"workflow": ..., "step": ...}) and carries the run’s user_id.
4. Closes the span with the result / exception.

Everywhere else (called directly, from a unit test, from an agent’s tool, or inside a workflow with neither telemetry nor an audit log), the decorator is transparent and the function runs with no overhead. That makes @step safe to leave on functions that are also called outside a workflow.

@step doesn’t journal the step, so a durable runtime never replays it: it runs again whenever it’s called.

Example

from loomflow import Agent, step
 
classifier = Agent("Classify the request.", model="gpt-4.1-mini")
billing    = Agent("Handle billing.",        model="gpt-4.1-mini")
tech       = Agent("Handle tech.",           model="gpt-4.1-mini")
 
@step
async def classify(text: str) -> str:
    return (await classifier.run(text)).output
 
@step
async def respond(text: str, label: str) -> str:
    handler = billing if label == "billing" else tech
    return (await handler.run(text)).output
 
# Plain Python control flow with a span on each step:
async def handle(text: str) -> str:
    label = await classify(text)
    return await respond(text, label)

Called directly (await handle("hello")), no spans fire. Run it as a workflow node with telemetry wired and the classify / respond spans appear in your trace, nested under the node’s own span:

from loomflow import Workflow
from loomflow.observability import InMemoryTelemetry
 
telemetry = InMemoryTelemetry()
wf = Workflow.chain([handle], telemetry=telemetry)
await wf.run("My invoice is wrong")
 
print([s.attributes["step"] for s in telemetry.spans()])
# ['classify', 'respond', 'handle']

Naming

The decorator preserves the wrapped function’s __name__ and __qualname__ so introspection helpers (Sphinx autodoc, debuggers) see the original identity. The wrapped function is also accessible via fn.__wrapped__ for unit tests that want to bypass the span and audit entries:

@step
async def classify(text: str) -> str:
    ...
 
# In a test:
result = await classify.__wrapped__("hello")    # skips span and audit

Custom step name

@step(name="ml-classifier-v2")
async def classify(text: str) -> str: ...

The span will be tagged step=ml-classifier-v2 regardless of the function’s __name__. Useful when:

• The function name is auto-generated or unstable (factory-built closures).
• You want to keep the trace name decoupled from refactors.
• You’re A/B-testing two variants under the same logical name.

@step or a Workflow node?

Every workflow node already gets its own span and its own step_started / step_completed audit entries (payload node=). @step adds them for the work inside a node (payload step=), so you can see which helper was slow or failed.

Telemetry attributes

Each loom.workflow.step span carries:

The span is a child of whatever span was active when the function ran, so a @step called from a workflow node nests under that node’s loom.workflow.step span.