Skip to main content

​
Definition

class Span(CamelModel):
    id: str
    trace_id: str
    parent_id: str | None
    name: str
    type: SpanType
    input: dict | None
    output: dict | None
    model: str | None
    start_time: int
    end_time: int | None
    duration: int | None
    token_usage: TokenUsage | None
    metadata: dict
    status: SpanStatus
    error: SpanError | None
    tags: list[str]
    user_id: str | None
    session_id: str | None
    version: str | None

​
Fields

​
id
string
Unique identifier for the span. Auto-generated UUID.
​
trace_id
string
required
ID of the parent trace.
​
parent_id
string
ID of the parent span for nested operations.
​
name
string
required
Name describing this operation.
​
type
SpanType
required
Type of operation: LLM, CHAIN, TOOL, or FUNCTION.
​
input
dict
Input data for this operation.
​
output
dict
Output data from this operation.
​
model
string
Model name for LLM spans (e.g., "claude-sonnet-4-6").
​
start_time
integer
Unix timestamp in milliseconds when the span started.
​
end_time
integer
Unix timestamp in milliseconds when the span ended.
​
duration
integer
Duration in milliseconds (end_time - start_time).
​
token_usage
TokenUsage
Token counts for LLM spans.
​
metadata
dict
Additional metadata (provider, parameters, etc.).
​
status
SpanStatus
Current status: RUNNING, SUCCESS, or ERROR.
​
error
SpanError
Error information if status is ERROR.
​
tags
list[string]
Tags inherited from the trace.
​
user_id
string
User identifier inherited from the trace.
​
session_id
string
Session identifier inherited from the trace.
​
version
string
Version string inherited from the trace.

​
SpanType Enum

class SpanType(str, Enum):
    LLM = "LLM"           # Language model calls
    CHAIN = "CHAIN"       # Orchestration logic
    TOOL = "TOOL"         # Tool/function execution
    FUNCTION = "FUNCTION" # Custom function spans

​
SpanStatus Enum

class SpanStatus(str, Enum):
    RUNNING = "RUNNING"   # In progress
    SUCCESS = "SUCCESS"   # Completed successfully
    ERROR = "ERROR"       # Failed with error

​
Methods

​
complete()

Mark span as successfully completed: 
span.complete(
    output={"result": "success"},
    token_usage=TokenUsage(
        prompt_tokens=100,
        completion_tokens=50,
        total_tokens=150
    )
)
token_usage is optional.

​
fail()

Mark span as failed: 
span.fail(
    error_message="API timeout",
    stack="Traceback (most recent call last)..."  # optional
)

​
Creating Custom Spans

Use the span() context manager — it handles complete() and fail() automatically: 
from rdk import observe, span
from rdk.models import SpanType

@observe()
def my_pipeline(query: str):
    with span("my-operation", span_type=SpanType.FUNCTION, input_data={"query": query}) as s:
        result = do_work(query)
        s.metadata["result_size"] = len(result)
    return result
See span() for full documentation.

​
JSON Serialization

{
  "id": "550e8400-e29b-41d4-a716-446655440001",
  "traceId": "550e8400-e29b-41d4-a716-446655440000",
  "parentId": null,
  "name": "anthropic.messages.create",
  "type": "LLM",
  "input": {"messages": [...]},
  "output": {"content": "..."},
  "model": "claude-sonnet-4-6",
  "startTime": 1708444800000,
  "endTime": 1708444801500,
  "duration": 1500,
  "tokenUsage": {
    "promptTokens": 100,
    "completionTokens": 50,
    "totalTokens": 150
  },
  "metadata": {"provider": "anthropic"},
  "status": "SUCCESS",
  "error": null,
  "tags": ["production"]
}

​
See Also