Using traces with Python - IBM watsonx Orchestrate ADK

Get traces from watsonx Orchestrate and export their spans by using a Python script.

"""
Example: Export and search trace data from Watson Orchestrate observability platform

This example demonstrates how to use the TracesController for programmatic access.
The controller is designed to be imported and used in custom Python scripts.

Prerequisites:
- Active Watson Orchestrate environment configured
- Admin access (traces endpoint requires admin privileges)
- Valid trace ID from your observability platform
"""

from ibm_watsonx_orchestrate.cli.commands.observability.traces.traces_controller import TracesController
from ibm_watsonx_orchestrate.client.base_api_client import ClientAPIException
from ibm_watsonx_orchestrate.client.observability.traces import TraceFilters, TraceSort, SpanCountRange
from datetime import datetime, timezone, timedelta

def example_basic_usage(trace_id):
    """
    Example 1: Fetch and analyze traces.
    
    """
    
    print("=" * 60)
    print("Example 1: Basic Usage")
    print("=" * 60)
    
    try:
        # Create controller
        controller = TracesController()
        
        # Fetch spans
        print(f"\nFetching spans for trace {trace_id}")
        spans_response = controller.fetch_trace_spans(trace_id)
        
        # Handle both response formats
        if spans_response.traceData:
            # New format: traceData with resourceSpans
            print(f"✓ Fetched trace data with resourceSpans")
            resource_spans = spans_response.traceData.resourceSpans
            print(f"  Resource spans: {len(resource_spans)}")
            
            # Count total spans across all resource spans
            total_spans = sum(
                len(scope_span.get('spans', []))
                for rs in resource_spans
                for scope_span in rs.get('scopeSpans', [])
            )
            print(f"  Total spans: {total_spans}")
            
            # Example: Find error spans in the raw data
            error_count = 0
            for rs in resource_spans:
                for scope_span in rs.get('scopeSpans', []):
                    for span in scope_span.get('spans', []):
                        status = span.get('status', {})
                        if status.get('code') == 'STATUS_CODE_ERROR':
                            error_count += 1
            print(f"  Error spans: {error_count}")
            
        elif spans_response.spans:
            # Legacy format: flat spans array
            print(f"✓ Fetched {len(spans_response.spans)} spans")
            print(f"  Total count: {spans_response.total_count}")
            
            # Analyze the spans (Python objects!)
            error_spans = [s for s in spans_response.spans
                          if s.status.status_code == "ERROR"]
            print(f"  Error spans: {len(error_spans)}")
        else:
            print("✗ No trace data found")
            return None
        
        return spans_response
        
    except ClientAPIException as e:
        print(f"✗ API Error ({e.response.status_code}): {e}")
        return None


def example_export_to_file(trace_id):
    """
    Example 2: Export specific trace to JSON file.
    """
    
    print("\n" + "=" * 60)
    print("Example 2: Export to JSON File")
    print("=" * 60)
    
    try:
        controller = TracesController()
        
        # Export to file (returns both objects and JSON string)
        print(f"\nExporting trace {trace_id} to file")
        spans_response, json_str = controller.export_trace_to_json(
            trace_id,
            output_file="my_trace.json",
            pretty=True
        )
        
        # Handle both response formats
        if spans_response.traceData:
            print(f"✓ Exported trace data to my_trace.json")
            print(f"  JSON string length: {len(json_str)} characters")
            print(f"  Format: traceData with resourceSpans")
        elif spans_response.spans:
            print(f"✓ Exported {len(spans_response.spans)} spans to my_trace.json")
            print(f"  JSON string length: {len(json_str)} characters")
            
            # You can still use the spans_response object
            for span in spans_response.spans[:3]:
                print(f"  - {span.name}: {span.status.status_code}")
        
        return spans_response
        
    except ClientAPIException as e:
        print(f"✗ API Error ({e.response.status_code}): {e}")
        return None


def example_custom_analysis(trace_id):
    """
    Example 3: Custom trace analysis.
    
    Shows how to use the controller for custom analysis workflows.
    """
    
    print("\n" + "=" * 60)
    print("Example 3: Custom Analysis")
    print("=" * 60)
    
    try:
        controller = TracesController()
        
        # Fetch spans
        print(f"\nAnalyzing trace {trace_id}")
        spans_response = controller.fetch_trace_spans(trace_id)
        
        # Custom analysis
        
        analysis = {
            'total_spans': 0,
            'errors': 0,
            'warnings': 0,
            'slow_spans': [],
            'span_types': {}
        }
        
        # Handle both response formats
        if spans_response.traceData:
            # Analyze traceData format
            for rs in spans_response.traceData.resourceSpans:
                for scope_span in rs.get('scopeSpans', []):
                    for span in scope_span.get('spans', []):
                        analysis['total_spans'] += 1
                        
                        # Count errors
                        status = span.get('status', {})
                        if status.get('code') == 'STATUS_CODE_ERROR':
                            analysis['errors'] += 1
                        
                        # Calculate duration
                        try:
                            start_nano = int(span.get('startTimeUnixNano', 0))
                            end_nano = int(span.get('endTimeUnixNano', 0))
                            duration_ms = (end_nano - start_nano) / 1_000_000
                            
                            if duration_ms > 1000:  # Slow spans (>1 second)
                                analysis['slow_spans'].append({
                                    'name': span.get('name', 'unknown'),
                                    'duration_ms': round(duration_ms, 2)
                                })
                        except:
                            pass
                        
                        # Count span types
                        kind = span.get('kind', 'UNKNOWN')
                        analysis['span_types'][kind] = analysis['span_types'].get(kind, 0) + 1
        
        elif spans_response.spans:
            # Analyze legacy spans format
            analysis['total_spans'] = len(spans_response.spans)
            
            for span in spans_response.spans:
                # Count errors
                if span.status.status_code == "ERROR":
                    analysis['errors'] += 1
                
                # Calculate duration
                try:
                    start = datetime.fromisoformat(span.start_time.replace('Z', '+00:00'))
                    end = datetime.fromisoformat(span.end_time.replace('Z', '+00:00'))
                    duration_ms = (end - start).total_seconds() * 1000
                    
                    if duration_ms > 1000:  # Slow spans (>1 second)
                        analysis['slow_spans'].append({
                            'name': span.name,
                            'duration_ms': round(duration_ms, 2)
                        })
                except:
                    pass
                
                # Count span types
                analysis['span_types'][span.kind] = analysis['span_types'].get(span.kind, 0) + 1
        
        # Print analysis
        print(f"\n✓ Analysis complete:")
        print(f"  Total spans: {analysis['total_spans']}")
        print(f"  Errors: {analysis['errors']}")
        print(f"  Slow spans (>1s): {len(analysis['slow_spans'])}")
        print(f"  Span types: {analysis['span_types']}")
        
        if analysis['slow_spans']:
            print(f"\n  Slowest spans:")
            for slow in sorted(analysis['slow_spans'], key=lambda x: x['duration_ms'], reverse=True)[:3]:
                print(f"    - {slow['name']}: {slow['duration_ms']}ms")
        
        return analysis
        
    except ClientAPIException as e:
        print(f"✗ API Error ({e.response.status_code}): {e}")
        return None


def example_search_traces():
    """
    Example 4: Search for traces using filters.
    
    Shows how to find trace IDs before exporting them.
    """
    print("\n" + "=" * 60)
    print("Example 4: Search for Traces")
    print("=" * 60)
    
    try:
        controller = TracesController()
        
        # Search for traces in the last 24 hours
        end_time = datetime.now(timezone.utc)
        start_time = end_time - timedelta(days=1)
        
        filters = TraceFilters(
            start_time=start_time.isoformat().replace('+00:00', 'Z'),
            end_time=end_time.isoformat().replace('+00:00', 'Z'),
            service_names=["wxo-server"]
        )
        
        sort = TraceSort(field="start_time", direction="desc")
        
        print(f"\nSearching for traces from {start_time.strftime('%Y-%m-%d %H:%M')} to {end_time.strftime('%Y-%m-%d %H:%M')}")
        search_response = controller.search_traces(
            filters=filters,
            sort=sort,
        )
        
        print(f"✓ Found {len(search_response.traceSummaries)} traces")
        
        if search_response.traceSummaries:
            print(f"\n  First 3 traces:")
            for trace in search_response.traceSummaries[:3]:
                print(f"    - Trace ID: {trace.traceId}")
                print(f"      Duration: {trace.durationMs}ms")
                print(f"      Spans: {trace.spanCount}")
                agent_name = trace.agentNames[0] if trace.agentNames else 'N/A'
                print(f"      Agent: {agent_name}")
        
        return search_response
        
    except ClientAPIException as e:
        print(f"✗ API Error ({e.response.status_code}): {e}")
        return None


def example_search_and_export():
    """
    Example 5: Search for traces, then export them.
    
    Complete workflow: search -> find trace IDs -> export error traces.
    """
    print("\n" + "=" * 60)
    print("Example 5: Search and Export Workflow")
    print("=" * 60)
    
    try:
        controller = TracesController()
        
        # Step 1: Search for traces with errors
        print("\nStep 1: Searching for traces with errors...")
        
        end_time = datetime.now(timezone.utc)
        start_time = end_time - timedelta(hours=1)
        
        filters = TraceFilters(
            start_time=start_time.isoformat().replace('+00:00', 'Z'),
            end_time=end_time.isoformat().replace('+00:00', 'Z')
        )
        
        search_response = controller.search_traces(
            filters=filters,
        )
        
        print(f"✓ Found {len(search_response.traceSummaries)} traces")
        
        # Step 2: Filter traces with errors (if root span has error status)
        error_traces = []
        for trace in search_response.traceSummaries:
            if trace.rootSpans:
                for root_span in trace.rootSpans:
                    if root_span.status.code == "STATUS_CODE_ERROR":
                        error_traces.append(trace)
                        break
        
        print(f"  Traces with errors: {len(error_traces)}")
        
        # Step 3: Export the first error trace
        if error_traces:
            trace_to_export = error_traces[0]
            print(f"\nStep 2: Exporting error trace {trace_to_export.traceId[:16]}...")
            
            spans_response, json_str = controller.export_trace_to_json(
                trace_to_export.traceId,
                output_file=f"error_trace_{trace_to_export.traceId[:8]}.json",
                pretty=True
            )
            
            if spans_response.traceData:
                print(f"✓ Exported trace data")
            elif spans_response.spans:
                print(f"✓ Exported {len(spans_response.spans)} spans")
            print(f"  File: error_trace_{trace_to_export.traceId[:8]}.json")
            
            return spans_response
        else:
            print("\n  No error traces found to export")
            return None
        
    except ClientAPIException as e:
        print(f"✗ API Error ({e.response.status_code}): {e}")
        return None




if __name__ == "__main__":
    # Example trace ID (replace with your actual trace ID)
    trace_id = "1234567890abcdef1234567890abcdef"
                
    print("\n" + "=" * 60)
    print("Watson Orchestrate Trace Export & Search Examples")
    print("=" * 60)
    print("\nThese examples show how to use TracesController")
    print("in your own Python scripts.")
    
    
    # Run export examples
    print("\n" + "=" * 60)
    print("PART 1: EXPORT EXAMPLES")
    print("=" * 60)
    example_basic_usage(trace_id)
    example_export_to_file(trace_id)
    example_custom_analysis(trace_id)
    
    # Run search examples
    print("\n" + "=" * 60)
    print("PART 2: SEARCH EXAMPLES")
    print("=" * 60)
    example_search_traces()
    example_search_and_export()
    
    print("\n" + "=" * 60)
    print("Examples completed!")
    print("=" * 60)
    print("\nKey points:")
    print("- Import TracesController from the CLI commands")
    print("- Controller methods return Python objects")
    print("- Use search_traces() to find trace IDs based on filters")
    print("- Use fetch_trace_spans() or export_trace_to_json() to get trace details")
    print("- Perfect for custom analysis, integrations, CI/CD")

from ibm_watsonx_orchestrate import Client, Credentials
        
  credentials = Credentials(url="<url>", api_key="<api_key>")
  client = Client(credentials)
        
  # Get all spans for a trace
  spans = client.service_instance.traces.get_spans(
      trace_id="1234567890abcdef1234567890abcdef"
  )

from datetime import datetime, timedelta

# Search for traces in the last 24 hours for specific service and agent name
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=1)

filters = TraceFilters(
    start_time=start_time.isoformat() + "Z",
    end_time=end_time.isoformat() + "Z",
    service_names=["wxo-server"],
    agent_names=["mobile-agent"]
)

sort = TraceSort(field="start_time", direction="desc")

response = client.service_instance.traces.search_traces(
    filters=filters,
    sort=sort,
)

print(f"Found {response.total_count} traces")
for trace in response.traces:
    print(f"Trace ID: {trace.trace_id}, Duration: {trace.duration_ms}ms")

References

Classes

Use these classes to handle trace data in the watsonx Orchestrate platform.

Import example

from ibm_watsonx_orchestrate.cli.commands.observability.traces.traces_controller import TracesController

TracesController

A controller that provides programmatic access to trace operations in Watson Orchestrate’s observability platform. Includes methods to search, fetch, and export traces, with optional pagination and CLI‑oriented progress logging.

Methods

get_client(self) -> TracesClient

Return the underlying TracesClient, creating it if necessary.Returns:

TracesClient

An authenticated client bound to the active environment.

fetch_trace_spans(self, trace_id: str, page_size: int = 100, fetch_all: bool = True, show_progress: bool = False) -> SpansResponse

Fetch spans for a specific trace IDParameters:

trace_id

string

A 32‑character hexadecimal trace ID.

page_size

int

Number of spans per page (1–1000). Default is 100.

fetch_all

bool

When True, retrieves spans across all pages. Default is True.

show_progress

bool

When True, logs progress through the logger. Default is False.

Returns:

SpansResponse

Contains spans and paging metadata. Depending on the platform:

Legacy: spans_response.spans (list of span objects) with total_count.
New (OTLP-style): spans_response.traceData.resourceSpans with scopeSpans[..].spans[..] (dicts).

export_trace_to_json(self, trace_id: str, output_file: Optional[str] = None, pretty: bool = True, page_size: int = 50) -> tuple[SpansResponse, str]

Fetch spans for a trace ID and export the data in JSON format. Uses TraceExporter for serialization and optionally writes to a file.Parameters:

trace_id

string

A 32‑character hexadecimal trace ID.

output_file

string (optional)

A file path for the JSON output. When None, JSON is returned as a string only. Default is None.

pretty

bool

Indented JSON for readability. Default is True.

page_size

int

Number of spans per page. Default is 50.

Returns:

tuple

SpansResponse: The spans fetched for programmatic use.
string: The JSON output, suitable for CLI display or storage.

search_traces(self, filters: Optional[TraceFilters] = None, sort: Optional[TraceSort] = None, page_size: int = 50, include_root_spans: bool = False, fetch_all: bool = True, show_progress: bool = False) -> TraceSearchResponse

Search traces using optional time filters and additional criteria such as service metadata, IDs, or span count ranges. Supports embedding root‑span data to quickly inspect error‑related attributes.Parameters:

filters

TraceFilters (optional)

Search criteria. Typically includes a time window:

start_time (str, RFC 3339 with Z)
end_time (str, RFC 3339 with Z) Optional fields:
service_names
agent_ids
agent_names
user_ids
session_ids
SpanCountRange Default is None.

sort

TraceSort (optional)

Sorting options (e.g., field=“start_time”, direction=“desc”). Default is None.

page_size

int

Results per page (1–100). Default is 100.

include_root_spans

bool

If True, embeds root span data in summaries (useful for filtering error traces by root_span.status.code). Default is False.

show_progress

bool

If True, logs progress messages via logger. Default is False.

Returns:

TraceSearchResponse

Contains traceSummaries and pagination details. Field names may vary depending on the model; traceSummaries is the typical structure.

Models

Use these models and the client to work with trace data in the watsonx Orchestrate platform.

Import example

from ibm_watsonx_orchestrate.client.observability.traces import TraceFilters, TraceSort, SpanCountRange

SpanContext

Span context containing trace and span identifier.Attributes

trace_id

string

A 32‑character hexadecimal trace ID.

span_id

string

A 16‑character hexadecimal span ID.

trace_state

string

Optional vendor‑specific trace metadata, such as W3C tracestate‑style key/value entries.

SpanStatus

Execution status of the span.Attributes

status_code

string

One of UNSET, OK, ERROR.

message

string

Optional human‑readable context about the status.

SpanEvent

Event that occurred during the span.Attributes

name

string

Event name.

timestamp

string

ISO 8601 timestamp in UTC.

attributes

object

Optional event attributes.

Span

OpenTelemetry‑compliant span object.Attributes

name

string

Human-readable operation name.

context

SpanContext

Trace and span identifiers.

parent_id

string

Parent span ID; None for root spans.

kind

string

Span kind such as INTERNAL, SERVER, or CLIENT.

start_time

string

ISO 8601 timestamp in UTC.

end_time

string

ISO 8601 timestamp in UTC.

status

SpanStatus

Execution status.

attributes

object

Arbitrary key-value attributes.

events

array

Events recorded during the span.

TraceData

Trace data containing resource spans.Attributes

resourceSpans

array

OpenTelemetry‑style structure, such as resourceSpans[].scopeSpans[].spans[*].

SpansResponse

Response from the get spans API. Supports two response shapes:

Structured: traceData.resourceSpans, an OpenTelemetry‑style grouping
Legacy: a flat spans array with pagination metadata

Attributes

traceData

TraceData

Structured trace data.

spans

array

Legacy spans array.

nextCursor

any

Pagination cursor using the internal camelCase field name.

totalCount

int

Reported total number of spans, using the internal camelCase field name.

SpanCountRange

Range used to filter traces by their span count.

TraceFilters

Filters used to search traces.Accepts ISO 8601 strings or Python datetime objects.
A field serializer converts datetime values to strings with a trailing Z.Attributes

start_time

string|datetime

Start time for filtering.

end_time

string|datetime

End time for filtering.

service_names

array

Service name filter. Supports multiple values and can be specified multiple times.

agent_ids

array

Agent ID filter. Allows selecting multiple agent identifiers and can be repeated as needed.

agent_names

array

Agent name filter. Accepts several agent names and may be provided multiple times.

user_ids

array

User ID filter. Supports filtering by one or more user identifiers, with repeatable entries.

session_ids

array

Session ID filter. Enables filtering by multiple session identifiers and can be included multiple times.

span_count_range

SpanCountRange

Thresholds for minimum and maximum span counts.

TraceSort

Sort configuration for trace searches.

TraceSearchRequest

Request body used to search traces.

RootSpanStatus

Status found in the root span, which may differ from the SpanStatus schema.

RootSpan

Root‑span data included in a trace summary.

TraceSummary

Summary data for a trace.

TraceSearchResponse

Response from the search‑traces API.Attributes

generatedAt

string

Time the response was generated.

originalQuery

object

Echoed search parameters.

traceSummaries

array

Summaries for each trace.

nextCursor

any

Pagination cursor.

totalCount

int

Total number of matching traces.

​References

​Classes

​Methods

​get_client(self) -> TracesClient

​fetch_trace_spans(self, trace_id: str, page_size: int = 100, fetch_all: bool = True, show_progress: bool = False) -> SpansResponse

​export_trace_to_json(self, trace_id: str, output_file: Optional[str] = None, pretty: bool = True, page_size: int = 50) -> tuple[SpansResponse, str]

​search_traces(self, filters: Optional[TraceFilters] = None, sort: Optional[TraceSort] = None, page_size: int = 50, include_root_spans: bool = False, fetch_all: bool = True, show_progress: bool = False) -> TraceSearchResponse

​Models

References

Classes

Methods

get_client(self) -> TracesClient

fetch_trace_spans(self, trace_id: str, page_size: int = 100, fetch_all: bool = True, show_progress: bool = False) -> SpansResponse

export_trace_to_json(self, trace_id: str, output_file: Optional[str] = None, pretty: bool = True, page_size: int = 50) -> tuple[SpansResponse, str]

search_traces(self, filters: Optional[TraceFilters] = None, sort: Optional[TraceSort] = None, page_size: int = 50, include_root_spans: bool = False, fetch_all: bool = True, show_progress: bool = False) -> TraceSearchResponse

Models