The @step decorator

from loomflow import step

Decorate any async def and it picks up telemetry + audit + journal hooks automatically. Control flow stays as if/await/return. You don’t have to learn a graph DSL just to get observability on a few agent calls.

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 inside a live RunContext (set by Workflow.run or by an enclosing Agent.run), the decorated function:

1. Reads the active telemetry from the context.
2. Opens a loom.workflow.step span tagged with step=<name>, user_id, session_id, pattern="workflow".
3. Executes the wrapped function inside the span.
4. Closes the span with the result / exception.

Outside any context, OR when the active telemetry is NoTelemetry, the decorator is transparent. The function runs with no overhead. This is the graceful-fallback property: @step is safe to leave on functions that may be called from a unit test or directly by a user.

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 + free observability:
async def my_workflow(text: str) -> str:
    label = await classify(text)
    return await respond(text, label)

When called via await my_workflow("hello") directly, no telemetry fires (no RunContext is live). When wrapped in a Workflow or an Agent’s tool dispatch, the per-step spans appear automatically in your trace.

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:

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

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.

When the decorator vs. when Workflow?

The two layer cleanly. @step is “I just want spans”; Workflow is “I want a structured graph the framework owns.”

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 inside a Workflow.run nests under loom.workflow.run, while a @step inside an Agent.run (called from a tool, say) nests under loom.run.