Skip to main content
GitHub

Decorators

Reference for all Risicare SDK decorators.

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

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",      # Role: orchestrator, worker, reviewer
    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:

ParameterTypeDefaultDescription
namestrFunction nameHuman-readable agent name (optional)
rolestrNoneAgent role (orchestrator, worker, reviewer)
agent_typestrNoneType identifier for filtering
versionintNoneAgent version number

Async Support:

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

Session Decorators

@session

Extracts session ID from function arguments and creates a session context.

from risicare import session
 
@session(session_id_arg="session_id", user_id_arg="user_id")
def handle_request(session_id: str, user_id: str, query: str):
    # All traces are grouped under this session
    return agent.run(query)

Parameters:

ParameterTypeDefaultDescription
session_id_argstr"session_id"Name of the function parameter containing session ID
user_id_argstr"user_id"Name of the function parameter containing user ID

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)

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:

ParameterTypeDefaultDescription
namestrFunction nameCustom span name
targetstrNoneTarget agent ID for the message recipient
target_namestrNoneHuman-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:

ParameterTypeDefaultDescription
namestrFunction nameCustom span name
targetstrNoneTarget agent ID that work is delegated to
target_namestrNoneHuman-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.

from risicare import trace
from risicare import SpanKind
 
@trace(
    name="custom-operation",
    kind=SpanKind.INTERNAL,
)
def custom_operation(data: dict):
    return process(data)

Parameters:

ParameterTypeDefaultDescription
namestrFunction nameSpan name
kindSpanKindINTERNALSpan kind
attributesdictNoneStatic 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