Python SDK

pip install risicare

risicare.init(
    api_key: str | None = None,
    endpoint: str | None = None,
    environment: str = "development",
    service_name: str | None = None,
    service_version: str | None = None,
    enabled: bool = True,
    trace_content: bool = True,
    sample_rate: float = 1.0,
    batch_size: int = 100,
    batch_timeout_ms: int = 1000,
    auto_patch: bool = True,
    debug: bool = False,
    exporters: list | None = None,
    metadata: dict | None = None,
    otlp_endpoint: str | None = None,
    otlp_headers: dict | None = None,
    otel_bridge: bool = False,
) -> RisicareClient

risicare.shutdown(timeout_ms: int = 5000) -> None

@risicare.agent(
    name: str | None = None,
    role: str | None = None,
    agent_type: str | None = None,
    version: int | None = None,
)
def my_agent():
    ...

@risicare.session(
    session_id_arg: str = "session_id",
    user_id_arg: str | None = "user_id",
)
def handle_request(session_id: str, user_id: str, query: str):
    ...

@risicare.trace_think
def analyze():
    ...

@risicare.trace_decide
def plan():
    ...

@risicare.trace_act
def execute():
    ...

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

@risicare.trace_message(
    name: str | None = None,
    target: str | None = None,
    target_name: str | None = None,
)
def send_to_reviewer(message: str):
    """Send a message to another agent."""
    return response

@risicare.trace_delegate(
    name: str | None = None,
    target: str | None = None,
    target_name: str | None = None,
)
def assign_research_task(task: str):
    """Delegate a research task to a worker agent."""
    return researcher.execute(task)

@risicare.trace_coordinate(
    name: str | None = None,
)
def sync_agents(agent_ids: list):
    """Coordinate state between multiple agents."""
    return sync_result

from risicare import SpanKind
 
@risicare.trace(
    name: str | None = None,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: dict | None = None,
)
def custom_operation():
    ...

with risicare.session_context(
    session_id: str,
    user_id: str | None = None,
    metadata: dict | None = None,
    parent_session_id: str | None = None,
    turn_number: int = 1,
):
    ...

with risicare.agent_context(
    agent_id: str,
    agent_name: str | None = None,
    agent_role: str | None = None,
    agent_type: str | None = None,
    version: str | None = None,
    metadata: dict | None = None,
):
    ...

from risicare import SemanticPhase
 
with risicare.phase_context(phase: SemanticPhase):
    # Example: phase_context(SemanticPhase.THINK)
    ...

risicare.is_enabled() -> bool

trace_id = risicare.get_current_trace_id() -> str | None

span_id = risicare.get_current_span_id() -> str | None

session_id = risicare.get_current_session_id() -> str | None

agent_id = risicare.get_current_agent_id() -> str | None

parent_span_id = risicare.get_current_parent_span_id() -> str | None

session = risicare.get_current_session() -> SessionContext | None

agent = risicare.get_current_agent() -> AgentContext | None

span = risicare.get_current_span() -> Span | None

phase = risicare.get_current_phase() -> SemanticPhase | None

ctx = risicare.get_current_context() -> TraceContext

trace_ctx = risicare.get_trace_context() -> TraceContext

tracer = risicare.get_tracer() -> Tracer

tracer = risicare.get_tracer()
span = tracer.start_span("my-operation")
span.set_attribute("key", "value")
span.end()

risicare.enable() -> None

risicare.disable() -> None

# Disable tracing for a sensitive operation
risicare.disable()
result = do_sensitive_work()
risicare.enable()

client = risicare.get_client() -> RisicareClient | None

risicare.register_span(span: Span) -> None

span = risicare.get_span_by_id(span_id: str) -> Span | None

# Useful in async generators where contextvars may be lost
async def stream_results(span_id: str):
    span = risicare.get_span_by_id(span_id)
    async for chunk in source:
        span.add_event("chunk", {"size": len(chunk)})
        yield chunk

risicare.unregister_span(span_id: str) -> None

risicare.traced_stream(
    span_id: str,
    stream: AsyncIterator,
    event_name: str = "chunk",
) -> AsyncIterator

span = tracer.start_span("stream-response")
async for chunk in risicare.traced_stream(span.span_id, response.stream()):
    print(chunk)

risicare.traced_stream_sync(
    span_id: str,
    stream: Iterator,
    event_name: str = "chunk",
) -> Iterator

span = tracer.start_span("stream-response")
for chunk in risicare.traced_stream_sync(span.span_id, response.stream()):
    print(chunk)

risicare.inject_trace_context(headers: dict) -> dict

headers = {"Content-Type": "application/json"}
risicare.inject_trace_context(headers)
# headers now includes "traceparent" and "tracestate"
response = httpx.post(url, headers=headers)

ctx = risicare.extract_trace_context(headers: dict) -> TraceContext | None

# In your HTTP handler
ctx = risicare.extract_trace_context(request.headers)
with risicare.restore_trace_context(ctx):
    process_request()

risicare.install_import_hooks() -> None

risicare.remove_import_hooks() -> None

risicare.instrument_already_imported() -> None

risicare.is_instrumented(module_name: str) -> bool

if risicare.is_instrumented("openai"):
    print("OpenAI is being traced")

modules = risicare.get_instrumented_modules() -> set[str]

supported = risicare.get_supported_modules() -> set[str]
# e.g. {"openai", "anthropic", "cohere", "langchain", ...}

from risicare import OTLPExporter
 
exporter = OTLPExporter(
    endpoint="http://localhost:4318/v1/traces",
    headers={"Authorization": "Bearer token"},
)
 
risicare.init(exporters=[exporter])

from risicare import FixRuntimeConfig
 
config = FixRuntimeConfig(
    enabled=True,
    poll_interval_s=30,
    cache_ttl_s=300,
)

risicare.init_runtime(config: FixRuntimeConfig) -> FixRuntime

runtime = risicare.get_runtime() -> FixRuntime | None

risicare.shutdown_runtime() -> None

async with risicare.async_session_context(
    session_id: str,
    user_id: str | None = None,
    metadata: dict | None = None,
    parent_session_id: str | None = None,
    turn_number: int = 1,
):
    await handle_request()

async with risicare.async_agent_context(
    agent_id: str,
    agent_name: str | None = None,
    agent_role: str | None = None,
    agent_type: str | None = None,
    version: str | None = None,
    metadata: dict | None = None,
):
    await run_agent()

with risicare.restore_trace_context(ctx: TraceContext):
    process_request()

from risicare import Span
 
span.trace_id       # str — 32 hex characters
span.span_id        # str — 16 hex characters
span.parent_span_id # str | None
span.name           # str
span.kind           # SpanKind
span.status         # SpanStatus
span.start_time     # float (epoch seconds)
span.end_time       # float | None
span.attributes     # dict
span.events         # list

from risicare import SpanKind
 
# Standard (OpenTelemetry compatible)
SpanKind.INTERNAL       # Default, internal operation
SpanKind.SERVER         # Server-side of RPC
SpanKind.CLIENT         # Client-side of RPC
SpanKind.PRODUCER       # Message producer
SpanKind.CONSUMER       # Message consumer
 
# Agent-specific
SpanKind.AGENT          # Agent lifecycle span
SpanKind.LLM_CALL       # LLM API call
SpanKind.TOOL_CALL      # Tool/function execution
SpanKind.RETRIEVAL      # RAG retrieval operation
SpanKind.DECISION       # Agent decision point
SpanKind.DELEGATION     # Delegation to another agent
SpanKind.COORDINATION   # Multi-agent coordination
SpanKind.MESSAGE        # Inter-agent message
 
# Phase-specific
SpanKind.THINK          # Reasoning/planning phase
SpanKind.DECIDE         # Decision-making phase
SpanKind.OBSERVE        # Environment observation phase
SpanKind.REFLECT        # Self-evaluation phase

from risicare import SpanStatus
 
SpanStatus.OK       # Successful
SpanStatus.ERROR    # Error occurred
SpanStatus.UNSET    # Status not set

from risicare import SemanticPhase
 
SemanticPhase.THINK       # Reasoning / analysis
SemanticPhase.DECIDE      # Decision making
SemanticPhase.ACT         # Action execution
SemanticPhase.REFLECT     # Self-evaluation
SemanticPhase.OBSERVE     # State reading
SemanticPhase.COMMUNICATE # Inter-agent messaging
SemanticPhase.COORDINATE  # Multi-agent coordination

from risicare import AgentRole
 
# Primary roles
AgentRole.ORCHESTRATOR   # Coordinates other agents
AgentRole.WORKER         # Executes assigned tasks
AgentRole.SUPERVISOR     # Monitors and validates
AgentRole.SPECIALIST     # Domain expert
 
# Communication roles
AgentRole.ROUTER         # Routes messages/tasks
AgentRole.AGGREGATOR     # Aggregates results
AgentRole.BROADCASTER    # Broadcasts to multiple agents
 
# Specialized roles
AgentRole.CRITIC         # Reviews and critiques
AgentRole.PLANNER        # Creates execution plans
AgentRole.EXECUTOR       # Executes plans
AgentRole.RETRIEVER      # Retrieves information
AgentRole.VALIDATOR      # Validates outputs

from risicare import MessageType
 
# Control messages
MessageType.TASK         # Task assignment
MessageType.RESULT       # Task result
MessageType.STATUS       # Status update
MessageType.ERROR        # Error notification
 
# Communication messages
MessageType.QUERY        # Information request
MessageType.RESPONSE     # Information response
MessageType.BROADCAST    # Broadcast to all
MessageType.DIRECT       # Direct message
 
# Coordination messages
MessageType.PROPOSAL     # Propose action
MessageType.VOTE         # Vote on proposal
MessageType.CONSENSUS    # Consensus reached
MessageType.CONFLICT     # Conflict detected
 
# Lifecycle messages
MessageType.HEARTBEAT    # Agent alive signal
MessageType.SHUTDOWN     # Shutdown signal
MessageType.HANDOFF      # Handoff to another agent

from risicare import SessionContext
 
session.session_id          # str
session.user_id             # str | None
session.metadata            # dict | None
session.parent_session_id   # str | None
session.turn_number         # int

from risicare import AgentContext
 
agent.agent_id    # str
agent.agent_name  # str | None
agent.agent_role  # str | None
agent.agent_type  # str | None
agent.version     # str | None
agent.metadata    # dict | None

from risicare import TraceContext
 
ctx.trace_id      # str
ctx.span_id       # str
ctx.trace_flags   # int
ctx.trace_state   # str | None