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 Agent Roles for all 12 values
    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, reviewer)
`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 the function name):

@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`	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`	Function name	Custom span name
`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 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 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:

from risicare import traced_stream
 
async for chunk in traced_stream(span_id=span.span_id, stream=llm.stream(prompt)):
    yield chunk

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