Decorators

Reference for all Risicare SDK decorators.

Risicare provides decorators for adding rich observability context to your agent code.

JavaScript SDK?

For JavaScript/TypeScript higher-order functions (agent, session, phases), see the JS Decorators guide.

Agent Decorators

@agent

Marks a function as an agent entry point. All spans created within the function are associated with this agent.

from risicare import agent
 
@agent(
    name="planner",           # Agent name (optional, defaults to function name)
    role="orchestrator",      # See AgentRole for all 12 values: /docs/reference/sdk#agentrole
    agent_type="custom",      # Type identifier
    version=1,                # Agent version number
)
def plan_task(objective: str):
    # All spans here are associated with this agent
    return result

Parameters:

Parameter	Type	Default	Description
`name`	`str`	Function name	Human-readable agent name (optional)
`role`	`str`	`None`	Agent role (orchestrator, worker, supervisor, critic — see `AgentRole` for all 12 values)
`agent_type`	`str`	`None`	Type identifier for filtering
`version`	`int`	`None`	Agent version number

Async Support:

@agent(name="async-planner")
async def async_plan_task(objective: str):
    result = await some_async_operation()
    return result

Context always propagates, even when tracing is off

@agent and phase decorators (@trace_think, @trace_decide, @trace_act, @trace_observe) always set their ContextVars regardless of whether tracing is enabled. This means get_current_agent() and get_current_phase() return values even when tracing is disabled. Only span creation is no-op'd — context propagation always runs.

Session Decorators

@session

Creates a session context. Can extract the session ID from function arguments, use a fixed ID, or auto-generate one.

from risicare import session
 
# Extract session_id from function argument (default behavior)
@session
def handle_request(session_id: str, user_id: str, query: str):
    return agent.run(query)
 
# Fixed session ID (no arg inspection needed)
@session(session_id="AAPL-analysis")
def analyze(ticker: str):
    return agent.run(ticker)
 
# Auto-generate a new UUID each call
@session(auto_generate=True)
def process(query: str):
    return agent.run(query)
 
# Custom argument names
@session(session_id_arg="sid", user_id_arg="uid")
def handle(sid: str, uid: str, query: str):
    return agent.run(query)

Parameters:

Parameter	Type	Default	Description
`session_id`	`str`	`None`	Fixed session ID (skips arg inspection)
`session_id_arg`	`str`	`"session_id"`	Name of the function parameter containing session ID
`user_id_arg`	`str`	`"user_id"`	Name of the function parameter containing user ID
`auto_generate`	`bool`	`False`	Generate a new UUID session ID per call

Session ID Resolution Priority

session_id kwarg (fixed value)
session_id_arg found in function arguments
auto_generate=True generates a fresh UUID
No match found — runs without session context

Phase Decorators

Track semantic decision phases:

@trace_think

Marks a function as a THINK phase (reasoning, planning).

from risicare import trace_think
 
@trace_think
def analyze_problem(context: dict):
    """Analyze the problem and identify key factors."""
    # Reasoning logic here
    return analysis

@trace_decide

Marks a function as a DECIDE phase (decision-making).

from risicare import trace_decide
 
@trace_decide
def choose_action(options: list):
    """Select the best action from available options."""
    # Decision logic here
    return selected_action

@trace_act

Marks a function as an ACT phase (action execution).

from risicare import trace_act
 
@trace_act
def execute_tool(tool_name: str, args: dict):
    """Execute a tool with the given arguments."""
    # Action execution here
    return result

@trace_observe

Marks a function as an OBSERVE phase (state reading).

from risicare import trace_observe
 
@trace_observe
def check_memory(key: str):
    """Read state from memory."""
    return memory.get(key)

All phase decorators accept an optional name parameter to customize the span name (defaults to <phase>:<function name>, for example think:analyze):

@trace_think(name="strategic-analysis")
def analyze(context: dict):
    ...

Multi-Agent Decorators

@trace_message

Tracks inter-agent message passing.

from risicare import trace_message
 
@trace_message
def send_to_reviewer(message: str):
    """Send a message to another agent."""
    return reviewer.receive(message)
 
@trace_message(target="reviewer-001", target_name="Code Reviewer")
def send_review_request(code: str):
    """Send a review request to the reviewer agent."""
    return reviewer.review(code)

Parameters:

Parameter	Type	Default	Description
`name`	`str`	`communicate:<function name>`	Custom span name
`target`	`str`	`None`	Target agent ID for the message recipient
`target_name`	`str`	`None`	Human-readable name of the target agent

@trace_delegate

Tracks task delegation to other agents.

from risicare import trace_delegate
 
@trace_delegate
def assign_research_task(task: str):
    """Delegate a research task to a worker agent."""
    return researcher.execute(task)
 
@trace_delegate(target="research-agent", target_name="Research Specialist")
def delegate_analysis(topic: str):
    """Delegate analysis to a specialist."""
    return specialist.analyze(topic)

Parameters:

Parameter	Type	Default	Description
`name`	`str`	`coordinate:<function name>`	Custom span name. A delegate span carries the `coordinate` phase
`target`	`str`	`None`	Target agent ID that work is delegated to
`target_name`	`str`	`None`	Human-readable name of the target agent

@trace_coordinate

Tracks coordination between multiple agents.

from risicare import trace_coordinate
 
@trace_coordinate
def sync_agents(agent_ids: list):
    """Coordinate state between multiple agents."""
    # Synchronization logic
    return sync_result

Generic Decorator

@trace

Creates a span with customizable kind and attributes. Works as both a decorator and a context manager.

As a decorator:

from risicare import trace
from risicare import SpanKind
 
# Bare decorator (uses function name as span name)
@trace
def process_data(data: dict):
    return transform(data)
 
# With parameters
@trace(name="custom-operation", kind=SpanKind.INTERNAL)
def custom_operation(data: dict):
    return process(data)

As a context manager:

from risicare import trace
 
def complex_pipeline(data):
    with trace("step-1"):
        intermediate = preprocess(data)
 
    with trace("step-2", attributes={"batch_size": len(data)}):
        result = transform(intermediate)
 
    return result

Parameters:

Parameter	Type	Default	Description
`name`	`str`	Function name	Span name
`kind`	`SpanKind`	`INTERNAL`	Span kind
`attributes`	`dict`	`None`	Static attributes to add to the span

Context Managers

For fine-grained control, use context managers instead of decorators:

agent_context

from risicare import agent_context, async_agent_context
 
# Sync
with agent_context("planner-001", agent_name="planner", agent_role="worker"):
    do_work()
 
# Async
async def main():
    async with async_agent_context("planner-001", agent_name="planner", agent_role="worker"):
        await do_work()

session_context

from risicare import session_context, async_session_context
 
# Sync
with session_context("sess-123", user_id="u-456"):
    agent.run(query)
 
# Async
async def main():
    async with async_session_context("sess-123", user_id="u-456"):
        await agent.run(query)

phase_context

from risicare import phase_context
from risicare import SemanticPhase
 
with phase_context(SemanticPhase.THINK):
    # This code is in THINK phase
    analyze()

Best Practices

Decorator Stacking

Decorators can be stacked. Apply them in order from outermost (first) to innermost (last):

@agent(name="planner")
@trace_think
def planning_phase():
    pass

Async Support

All decorators support both sync and async functions automatically.

Generator Functions

For generator functions, use the streaming utilities instead of decorators — see Streaming:

from risicare import traced_stream, get_tracer
 
async def generate(prompt):
    with get_tracer().start_span("llm_stream") as span:
        async for chunk in traced_stream(span_id=span.span_id, stream=llm.stream(prompt)):
            yield chunk

Read a generator like this to its end, or close it with contextlib.aclosing() where you stop early. A generator that is dropped part-way and closed later, in another task or thread, can leave its ended span as the current span, so the next span becomes its child (a known issue that the release does not fix).

Next Steps

Context Propagation

How context flows through your code

Learn more

Multi-Agent Observability

Track agent interactions

Learn more

Edit this page on GitHub

PreviousConfiguration NextContext