Skip to main content

​
Signature

def observe(
    name: str | None = None,
    *,
    tags: list[str] | None = None,
    metadata: dict | None = None,
    user_id: str | Callable[[], str | None] | None = None,
    session_id: str | Callable[[], str | None] | None = None,
    version: str | None = None,
    trace_id: str | Callable[[], str | None] | None = None,
    capture_input: bool = False,
    capture_output: bool = False,
    tool_mocks: ToolMocks | None = None,
    tool_mock_default: Any = USE_GLOBAL_DEFAULT,
) -> Callable

​
Parameters

​
name
string
default:"function name"
Name for the trace. Defaults to the decorated function’s name if not provided.Example: "chat-completion", "customer-support", "process-order"
​
tags
list[string]
default:"[]"
Tags to attach to all spans in this trace.Example: ["production", "high-priority"]
​
metadata
dict
default:"{}"
Custom key-value pairs to attach to the trace.Example: {"order_id": "123", "plan": "enterprise"}
​
user_id
string | Callable[[], str | None]
default:"None"
User identifier. Can be a static string or a zero-argument callable that returns a string.Example: "user_123" or lambda: get_current_user_id()
​
session_id
string | Callable[[], str | None]
default:"None"
Session identifier for grouping related traces. Accepts a string or callable.
​
version
string
default:"None"
Version string stored in trace metadata (e.g., model version, app version).Example: "1.0.0", "claude-sonnet-4-6"
​
trace_id
string | Callable[[], str | None]
default:"auto-generated"
Explicit trace ID. If not provided, a UUID is generated. Accepts a string or callable.Use this to correlate traces across services.
​
capture_input
boolean
default:false
If True, serialize the function’s arguments and store them in trace.metadata["input"].
​
capture_output
boolean
default:false
If True, serialize the function’s return value and store it in trace.metadata["output"].
​
tool_mocks
ToolMocks
default:"None"
Mapping of tool names to mock outputs or callables. Active only within this function call.See mock_tools() for details.
​
tool_mock_default
Any
default:"USE_GLOBAL_DEFAULT"
Default mock response for any tool not in tool_mocks. Falls back to the process-wide default set by test_mode.

​
Examples

​
Basic

from rdk import observe

@observe()
def chat(message: str) -> str:
    # trace name defaults to "chat"
    ...

@observe(name="customer-support")
def handle_request(message: str) -> str:
    ...

​
With metadata

from rdk import observe

@observe(
    name="chat-pipeline",
    tags=["production", "claude"],
    metadata={"team": "ml-platform"},
    version="2.1.0",
)
def pipeline(question: str) -> str:
    ...

​
Dynamic user and session

from rdk import observe

@observe(
    name="chat",
    user_id=lambda: get_current_user_id(),    # Callable — evaluated at call time
    session_id=lambda: get_session_id(),
)
def chat(message: str) -> str:
    ...

​
Capture inputs and outputs

from rdk import observe

@observe(name="summarize", capture_input=True, capture_output=True)
def summarize(text: str) -> str:
    # Arguments stored in trace.metadata["input"]
    # Return value stored in trace.metadata["output"]
    ...

​
Sync and async

Works with both sync and async functions: 
from rdk import observe

@observe(name="sync-chat")
def sync_chat(message: str) -> str:
    ...

@observe(name="async-chat")
async def async_chat(message: str) -> str:
    ...

​
Nesting

Nested @observe decorators skip the inner trace — only the outermost creates a trace context. All LLM calls within nested functions are attributed to the outermost trace. 
@observe(name="outer")
def outer():
    inner()  # No new trace created; spans go to "outer"

@observe(name="inner")
def inner():
    ...

​
Context manager alternative

If decorators aren’t convenient (notebooks, scripts), use trace(): 
from rdk import trace

with trace("my-operation", tags=["dev"]) as t:
    print(f"Trace ID: {t.id}")
    # LLM calls here are captured
See trace() for details.

​
See Also