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 resultParameters:
| 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 resultSession 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:
| Parameter | Type | Default | Description |
|---|---|---|---|
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 |
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:
| 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_resultGeneric 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:
| 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():
passAsync 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