Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.flowx.ai/llms.txt

Use this file to discover all available pages before exploring further.

Three decorators cover the vast majority of instrumentation needs. Each captures inputs and outputs, measures duration, and emits *_start and *_end events to Observatory.

​
@agent

The top-level decorator. Wrap the function that handles one user request. 
from flowxobservatory import agent

@agent("claims-approval")
def run(ctx, payload):
    # business logic
    return result

​
Parameters

​
name
string
required
The agent identifier. Shows up in the Agent column on LLM Calls, in Risk Dashboard, and in ROI baselines. Use kebab-case.
​
user_id
string
Identifier of the user this run belongs to. Surfaces in filters and Threads grouping.
​
user_props
dict
Arbitrary user properties attached to the run (e.g. {"plan": "enterprise"}).
​
tags
list[str]
Tags attached to every run of this agent (e.g. ["claims", "prod"]). Used for filtering in LLM Calls and analytics.

​
@chain

Use for sub-steps that aren’t external work but are worth seeing in the trace. 
from flowxobservatory import chain

@chain("extract-claim-data")
def extract(payload):
    return {"holder": payload["policy_id"], ...}

​
Parameters

Same shape as @agent. The chain inherits the parent run’s app and project context — you only need to name it.
If you’re using LangChain or LangGraph, the callback handler creates chain spans automatically. Use @chain for handwritten Python that lives between LangChain calls or for non-LangChain frameworks.

​
@tool

Use for the leaf-level units of work — database queries, REST calls, file operations, computation. 
from flowxobservatory import tool

@tool("lookup_policy_holder")
def lookup_policy_holder(policy_id):
    return db.fetch(policy_id)

​
Parameters

Same signature as @agent — name, user_id, user_props, tags.

​
Async functions

All three decorators detect async def and wrap it natively. No different syntax. 
@agent("claims-approval")
async def run(ctx, payload):
    data = await extract(payload)
    return await decide(data)

​
Error handling

If the wrapped function raises, the SDK:
  1. Emits an *_end event with status=error and the exception’s class and message.
  2. Re-raises the exception unchanged.
Your error handling does not need to change. Observatory captures the error context; your try/except still runs.

​
Class-based instrumentation

The decorators also work on methods. Use @agent on the entry method of a class: 
class ClaimsAgent:
    @agent("claims-approval")
    def run(self, ctx, payload):
        ...
self is excluded from captured inputs automatically.

​
When NOT to use a decorator

SituationBetter choice
You want to attach a custom event (saved, chat)Call track_event() directly.
The function runs millions of times per requestDecorate the parent, not the leaf.
You only want to record success metricsUse the analytics API on the recorded run instead.

Event reporting

Custom events and the wire protocol.

SDK overview

Install and configuration.
Last modified on June 2, 2026