Python SDK

Reference for the Risicare Python SDK. The current published version is 0.6.0 — see the Changelog.

Installation

pip install risicare

Exports

The risicare package exports 84 public names (risicare.__all__). The tables below list them by category:

Core

Tracer

Decorators

Context Managers

Context Accessors

Span Registry

Streaming

W3C Trace Context

Types

Fix Runtime

Auto-Instrumentation

Exporters

Masking

Webhooks

risicare.init

Initialize the SDK.

risicare.init(
    api_key: str | None = None,
    *,
    endpoint: str | None = None,
    api_endpoint: str | None = None,
    project_id: str | None = None,
    environment: str = "development",
    service_name: str | None = None,
    service_version: str | None = None,
    enabled: bool = True,
    trace_content: bool = False,
    sample_rate: float = 1.0,
    batch_size: int = 500,
    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,
    mask: Callable[[str, Any], Any] | None = None,
    fix_runtime: bool = False,
) -> RisicareClient

Every argument after api_key is keyword-only: risicare.init("rsk-...", "https://...") raises TypeError.

risicare.shutdown

Graceful shutdown — flushes pending spans and closes connections.

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

The shutdown sequence:

1. Signals the batch processor to stop
2. Waits up to timeout_ms for the worker thread to drain its queue
3. Performs a final flush of any remaining buffered spans
4. Shuts down all exporters (HTTP connections closed)
5. Restores the executor and asyncio patches if auto_patch was enabled. Provider patches stay

risicare.flush

Send the spans queued so far. The SDK stays usable after the call.

risicare.flush(timeout_ms: int = 5000) -> bool

Returns a bool. True means that the gateway (or an exporter that you named) acknowledged every span and every score that the SDK accepted since the previous flush() returned (or since init()), and that nothing is in the queue or in flight. This holds also in a signal handler and after shutdown(). A span that you start after shutdown() has returned is not made, and a call to a traced provider in your handler after the SDK's drain makes no span; nothing counts them, so a True does not show that they were traced.

A loss makes the next flush() return False one time, and the one after it True again. A loss is a span dropped for any reason (shutdown_residue included), a span rejected or left out of an acknowledgement, and a failed score (also a score that score() refuses at the call). Each one raises a counter that names the reason: dropped_spans_by_reason, rejected_spans, failed_scores. A flush() that returns False at its deadline while none of these counters went up since the previous flush() means that the spans are still in flight, not lost. failed_exports is not a loss counter: a later round can still deliver those spans.

flush() also returns False when the SDK has no evidence of delivery: before init(), with tracing off, and with no API key (and no exporter of your own). A flush() that overlaps the SDK's own SIGTERM drain, or a shutdown() on another thread, waits for it inside its own deadline. With more than one exporter, a batch counts as delivered when the first exporter that succeeds takes it, so True means that at least one of them acknowledged each span. A new init() starts a new count: report a loss with flush() or get_metrics() before you call init() again.

flush() only waits: it does not stop a request that is already sent. To end a process in a fixed time, call shutdown(timeout_ms). When flush() returns False, read risicare.get_metrics().

In Python 0.5.1 and earlier, flush() returned True before init(), with tracing off and with no key, and it could return True in a signal handler or after shutdown() with nothing delivered. After a shutdown() that left spans, it returned False for every later call.

risicare.get_metrics

Return the SDK's export counters.

risicare.get_metrics() -> dict

import risicare
 
risicare.init(api_key="rsk-...")
# ... your application code ...
risicare.flush()
metrics = risicare.get_metrics()
print(metrics["exported_spans"], metrics["failed_exports"], metrics["dropped_spans"])

A gateway acknowledgement means that the spans are in the gateway's queue. It does not mean that they are stored. To confirm that a trace is stored, open it in the dashboard.

@risicare.agent

Decorator for agent functions.

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

@risicare.session

Decorator for session-bound functions.

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

@risicare.trace_think

Decorator for thinking phase.

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

@risicare.trace_decide

Decorator for decision phase.

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

@risicare.trace_act

Decorator for action phase.

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

@risicare.trace_observe

Decorator for observation phase (state reading).

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

@risicare.trace_message

Decorator for inter-agent message passing.

@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

Decorator for task delegation.

@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

Decorator for multi-agent coordination.

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

@risicare.trace

Generic trace decorator with customizable kind.

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

risicare.session_context

Context manager for sessions.

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,
):
    ...

risicare.agent_context

Context manager for agents.

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

risicare.phase_context

Context manager for phases.

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

risicare.is_enabled

Check if tracing is enabled. It is False before init(), with tracing off, with no API key (and no exporter of your own), and after the SDK's own SIGTERM drain. In 0.5.1 and earlier it was True with no key.

risicare.is_enabled() -> bool

risicare.get_current_trace_id

Get current trace ID.

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

risicare.get_current_span_id

Get current span ID.

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

risicare.get_current_session_id

Get current session ID.

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

risicare.get_current_agent_id

Get current agent ID.

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

risicare.get_current_parent_span_id

Get the parent span ID from the current context.

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

risicare.get_current_session

Get the active SessionContext, or None if no session is active.

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

Returns the full SessionContext dataclass with session_id, user_id, metadata, parent_session_id, and turn_number.

risicare.get_current_agent

Get the active AgentContext, or None if no agent context is active.

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

Returns the full AgentContext dataclass with agent_id, agent_name, agent_role, agent_type, version, and metadata.

risicare.get_current_span

Get the current active span, or None if no span is active.

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

risicare.get_current_phase

Get the active SemanticPhase, or None if no phase is active.

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

Returns one of: SemanticPhase.THINK, DECIDE, ACT, REFLECT, OBSERVE, COMMUNICATE, COORDINATE.

risicare.get_current_context

Get the full trace context state.

ctx = risicare.get_current_context() -> dict

Returns a dict with session, agent, span, and phase keys (each None when not set), capturing the current context state in a single call. Useful for debugging, logging, or serialization.

risicare.get_trace_context

Get W3C trace context for distributed tracing propagation.

trace_ctx = risicare.get_trace_context() -> TraceContext

risicare.get_tracer

Get the global Tracer instance. Use this for low-level span creation.

tracer = risicare.get_tracer() -> Tracer

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

risicare.report_error

Report a caught exception so Risicare records it on the trace and classifies it against the error taxonomy. Use this when your code catches an exception you still want recorded. Root-cause analysis and fix suggestions are held for the beta, so reporting an error does not produce them.

risicare.report_error(
    exception: BaseException,
    *,
    name: Optional[str] = None,
    attributes: Optional[Dict[str, Any]] = None,
) -> None

Inside a traced context: records the exception on the current span and sets error=True, error.type, and error.message attributes. The span's status is left unchanged — the traced function may still complete successfully.

Outside any traced context: creates a standalone error span with status ERROR. Duplicate errors (same type + message) are suppressed for 5 minutes.

This function never raises.

from risicare import report_error
 
def answer(prompt):
    try:
        return llm.invoke(prompt)
    except openai.RateLimitError as e:
        report_error(e)          # Records the error on a span and classifies it
        return cached_response

risicare.score

Record a custom evaluation score on a trace. Scores appear in the dashboard and can be used for filtering, analytics, and quality monitoring.

risicare.score(
    trace_id: str,
    name: str,
    value: float,
    *,
    span_id: Optional[str] = None,
    comment: Optional[str] = None,
) -> None

This function is non-blocking — it dispatches the score in a background thread and returns immediately. It never raises. A failed request (an error, a timeout, or a response status of 300 or more) adds 1 to failed_scores, and the next flush() returns False. The first failure logs a WARNING with the score name and the reason; failures in the next 10 seconds are counted, and the next WARNING gives their number.

import risicare
 
risicare.init(api_key="rsk-...")
 
risicare.score(
    trace_id="4bf92f3577b34da6a3ce929d0e0e4736",
    name="factual_accuracy",
    value=0.92,
    comment="Verified against source documents"
)

risicare.enable

Re-enable tracing after it was disabled. It cannot turn tracing on when the SDK has no API key (and no exporter of your own): is_enabled() stays False.

risicare.enable() -> None

risicare.disable

Temporarily disable all tracing. Instrumented calls become no-ops with zero overhead.

risicare.disable() -> None

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

risicare.get_client

Get the current RisicareClient instance, or None if not initialized.

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

Span Registry

The span registry provides ID-based span lookup, which is essential for contexts where contextvars may not propagate reliably (async generators, thread pools).

risicare.register_span

Register a span in the global registry for later retrieval by ID.

risicare.register_span(
    span: Span,
    ttl_seconds: Optional[float] = None,
) -> None

A registered span expires after ttl_seconds (default: 60 seconds). The SDK registers each span while it is open, also the spans of decorators and context managers, and removes it when the span ends. The TTL applies to an open span too: for a span that stays open longer, call extend_span_ttl(), which risicare exports since 0.5.1 (see Streaming). Use register_span() to give a span a different TTL.

risicare.get_span_by_id

Retrieve a span by its ID from the global registry.

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

Remove a span from the global registry.

risicare.unregister_span(span_id: str) -> None

A span leaves the registry when it ends, so this call is optional.

Streaming

Utilities for tracing streaming responses from LLM providers.

risicare.traced_stream

Wrap an async iterator with tracing. Each chunk adds an event to the span, and the totals are set when the stream ends. Use it inside a with tracer.start_span() block: the block ends the span, also when the code raises before the stream starts. Outside a block, traced_stream ends the span itself when the stream completes, raises or is closed (status unset); a stream that never starts then leaves its span open and unsent. When the block is inside an async generator, read the generator to its end or close it with contextlib.aclosing() — see Streaming.

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

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

risicare.traced_stream_sync

Synchronous version of traced_stream.

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

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

W3C Trace Context

Functions for propagating trace context across service boundaries using W3C traceparent and tracestate headers.

risicare.inject_trace_context

Inject W3C traceparent and tracestate headers into a headers dict.

risicare.inject_trace_context(headers: dict) -> dict

headers = {"Content-Type": "application/json"}
risicare.inject_trace_context(headers)
# Inside an active span, headers now includes "traceparent".
# "tracestate" is added only when a session or an agent is current.
response = httpx.post(url, headers=headers)

risicare.extract_trace_context

Extract trace context from incoming W3C headers.

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

Returns a dict (empty if no trace headers are present). When a valid traceparent is present it always includes version, trace_id, parent_span_id, and flags; session_id and agent_id are added only when the caller propagated them via tracestate. To restore context, build a TraceContext from these keys first — restore_trace_context() expects a TraceContext object, not the raw dict:

from risicare import TraceContext, restore_trace_context, extract_trace_context
 
# In your HTTP handler. The lookup needs lower-case header names.
raw = extract_trace_context({k.lower(): v for k, v in request.headers.items()})
if raw.get("trace_id") and raw.get("parent_span_id"):
    ctx = TraceContext(
        trace_id=raw["trace_id"],
        span_id=raw["parent_span_id"],
        session_id=raw.get("session_id"),
        agent_id=raw.get("agent_id"),
    )
    with restore_trace_context(ctx):
        process_request()

Auto-Instrumentation

Functions to control which modules are auto-instrumented.

risicare.install_import_hooks

Install Python import hooks for auto-instrumentation. Called automatically by risicare.init() when tracing is enabled, whatever the value of auto_patch.

risicare.install_import_hooks(
    allowed_modules: Optional[Set[str]] = None,
) -> RisicareImportFinder

Pass allowed_modules to limit which modules are instrumented (e.g., {"openai", "anthropic"}). Returns the import finder instance.

risicare.remove_import_hooks

Remove auto-instrumentation import hooks. Modules already instrumented remain instrumented.

risicare.remove_import_hooks() -> None

risicare.instrument_already_imported

Instrument modules that were imported before risicare.init() was called. Called automatically by init().

risicare.instrument_already_imported() -> int

Returns the number of modules that were instrumented.

risicare.is_instrumented

Check if at least one instrumentation hook landed for a module. True when hooks_landed > 0, also for a partial patch. False when no hook landed, when the module is pending, and when it was never imported. For the full state, read get_instrumented_modules().

risicare.is_instrumented(module_name: str) -> bool

if risicare.is_instrumented("openai"):
    print("OpenAI calls are traced")

risicare.get_instrumented_modules

Get the per-module instrumentation state. Returns a dict keyed by module name; each value is a state record with eight keys: state, hooks_landed, hooks_attempted, succeeded, failed, inert, version_compat and targets_resolved.

modules = risicare.get_instrumented_modules() -> dict[str, dict]
# {
#   "openai":    {"state": "instrumented", "hooks_landed": 4, ...},
#   "langchain": {"state": "attached-inert", "hooks_landed": 0, ...},
# }

state is one of instrumented, attached-inert, partially_instrumented, failed, pending or attempted. Only state == "instrumented" means the integration is attached and on a live call path. A patched provider reports instrumented, and one whose patch raised reports failed. pending: the module is imported, but the module that holds its patch points is not yet (for example botocore imported and botocore.client not yet, or vertexai before vertexai.generative_models). The OpenTelemetry bridge still reports attempted.

risicare.get_supported_modules

Get the set of all module names supported for auto-instrumentation.

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

Exporters

OTLPExporter

Export spans via OTLP/HTTP protocol. Use this when sending traces to an OpenTelemetry-compatible backend alongside Risicare.

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

Fix Runtime

The fix runtime loads, caches, and applies fixes at the SDK level. It runs inside your agent process and intercepts matching errors to apply the configured fix strategy.

FixRuntimeConfig

Configuration for the fix runtime.

from risicare import FixRuntimeConfig
 
config = FixRuntimeConfig(
    enabled=True,
    refresh_interval_seconds=30,
    cache_ttl_seconds=300,
)

See Fix Runtime for every field and its default.

risicare.init_runtime

Initialize the fix runtime with a configuration.

risicare.init_runtime(config: Optional[FixRuntimeConfig] = None, auto_start: bool = True) -> FixRuntime

risicare.get_runtime

Get the current fix runtime instance.

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

risicare.shutdown_runtime

Shutdown the fix runtime gracefully.

risicare.shutdown_runtime() -> None

FixRuntime

Main fix runtime class. Coordinates FixLoader, FixCache, FixApplier, and FixInterceptor.

FixLoader

Loads fix configurations from the Risicare API.

FixCache

Caches fix configurations locally with configurable TTL to minimize API calls.

FixApplier

Applies a fix configuration to an error span (e.g., retry, fallback, parameter change).

FixInterceptor

Intercepts errors matching a fix rule and applies the configured fix before the error propagates.

Async Context Managers

risicare.async_session_context

Async version of session_context for use in async code.

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()

risicare.async_agent_context

Async version of agent_context for use in async code.

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: int | None = None,
    metadata: dict | None = None,
):
    await run_agent()

risicare.restore_trace_context

Restore a previously saved or extracted TraceContext. Use this to re-establish context in a new scope, such as after extracting from W3C headers or restoring in a background task.

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

Types

Span

Core span data class representing a single unit of work.

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

SpanKind

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

SpanStatus

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

SemanticPhase

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

AgentRole

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

MessageType

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

ExceptionInfo

Exception data attached to spans. Imported from risicare_core (not re-exported by the risicare package).

from risicare_core import ExceptionInfo
 
info = ExceptionInfo(
    type="TimeoutError",
    message="Request timed out after 30s",
    stacktrace="Traceback (most recent call last):\n  ...",
    # timestamp: auto-set to UTC now
    # escaped: defaults to True
)

SessionContext

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

AgentContext

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

TraceContext

from risicare import TraceContext
 
ctx.trace_id      # str — 32-char lowercase hex
ctx.span_id       # str — 16-char lowercase hex
ctx.session_id    # str | None
ctx.agent_id      # str | None

Testing Utilities

risicare.reset_for_testing

Reset all SDK state for clean test isolation. Call in test teardown to prevent state leakage between tests.

risicare.reset_for_testing() -> None

Performs a full reset:

1. Calls shutdown(timeout_ms=5000) (swallows errors)
2. Resets the client singleton
3. Resets the tracer singleton
4. Resets all context variables (session, agent, span, phase)

# pytest fixture
import pytest
import risicare
 
@pytest.fixture(autouse=True)
def clean_risicare():
    yield
    risicare.reset_for_testing()

Next Steps