> ## Documentation Index
> Fetch the complete documentation index at: https://developer.watson-orchestrate.ibm.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Using traces with Python

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

<CodeGroup>
  ```python Fetch and analyze observations [expandable] theme={null}
  """
  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
  from datetime import datetime, timezone, timedelta


  def example_basic_usage(trace_id):
      """
      Example 1: Fetch and analyze observations for a trace.
      """

      print("=" * 60)
      print("Example 1: Basic Usage")
      print("=" * 60)

      try:
          controller = TracesController()

          print(f"\nFetching observations for trace {trace_id}")
          obs_response = controller.fetch_trace_observations(trace_id)

          if obs_response.observations:
              print(f"✓ Fetched {len(obs_response.observations)} observations")
              print(f"  Total count: {obs_response.total_count}")

              # Analyse the observations
              generations = [o for o in obs_response.observations if o.type == "GENERATION"]
              print(f"  LLM calls (GENERATION): {len(generations)}")
          else:
              print("✗ No observations found")
              return None

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

          print(f"\nExporting trace {trace_id} to file")
          obs_response, json_str = controller.export_trace_to_json(
              trace_id,
              output_file="my_trace.json",
              pretty=True
          )

          if obs_response.observations:
              print(f"✓ Exported {len(obs_response.observations)} observations to my_trace.json")
              print(f"  JSON string length: {len(json_str)} characters")

          return obs_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.
      """

      print("\n" + "=" * 60)
      print("Example 3: Custom Analysis")
      print("=" * 60)

      try:
          controller = TracesController()

          print(f"\nAnalyzing trace {trace_id}")
          obs_response = controller.fetch_trace_observations(trace_id)

          analysis = {
              'total_observations': 0,
              'by_type': {},
              'slow_observations': [],
          }

          if obs_response.observations:
              analysis['total_observations'] = len(obs_response.observations)

              for obs in obs_response.observations:
                  # Count by type
                  analysis['by_type'][obs.type] = analysis['by_type'].get(obs.type, 0) + 1

                  # Find slow observations (> 1 second)
                  try:
                      start = datetime.fromisoformat(obs.startTime.replace('Z', '+00:00'))
                      end = datetime.fromisoformat(obs.endTime.replace('Z', '+00:00'))
                      duration_ms = (end - start).total_seconds() * 1000

                      if duration_ms > 1000:
                          analysis['slow_observations'].append({
                              'name': obs.name,
                              'duration_ms': round(duration_ms, 2)
                          })
                  except Exception:
                      pass

          print(f"\n✓ Analysis complete:")
          print(f"  Total observations: {analysis['total_observations']}")
          print(f"  By type: {analysis['by_type']}")
          print(f"  Slow observations (>1s): {len(analysis['slow_observations'])}")

          if analysis['slow_observations']:
              print(f"\n  Slowest observations:")
              for slow in sorted(analysis['slow_observations'], 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.
      """
      print("\n" + "=" * 60)
      print("Example 4: Search for Traces")
      print("=" * 60)

      try:
          controller = TracesController()

          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'),
          )

          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")
                  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.
      """
      print("\n" + "=" * 60)
      print("Example 5: Search and Export Workflow")
      print("=" * 60)

      try:
          controller = TracesController()

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

          print("\nStep 1: Searching for recent traces...")
          search_response = controller.search_traces(filters=filters)
          print(f"✓ Found {len(search_response.traceSummaries)} traces")

          if search_response.traceSummaries:
              trace_to_export = search_response.traceSummaries[0]
              print(f"\nStep 2: Exporting trace {trace_to_export.traceId[:16]}...")

              obs_response, json_str = controller.export_trace_to_json(
                  trace_to_export.traceId,
                  output_file=f"trace_{trace_to_export.traceId[:8]}.json",
                  pretty=True
              )

              if obs_response.observations:
                  print(f"✓ Exported {len(obs_response.observations)} observations")
              print(f"  File: trace_{trace_to_export.traceId[:8]}.json")

              return obs_response

          print("\n  No 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)

      # 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_observations() or export_trace_to_json() to get trace details")
      print("- Perfect for custom analysis, integrations, CI/CD")
  ```
</CodeGroup>

## References

### Classes

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

```python Import example theme={null}
from ibm_watsonx_orchestrate.cli.commands.observability.traces.traces_controller import TracesController
```

<AccordionGroup>
  <Accordion title="TracesController">
    Use `TracesController` to access trace operations programmatically in the watsonx Orchestrate observability platform.
    It exposes 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:**

    <ResponseField name="TracesClient">
      An authenticated client bound to the active environment.
    </ResponseField>

    ### fetch\_trace\_observations(self, trace\_id: str, page\_size: int = 100, fetch\_all: bool = True, show\_progress: bool = False) -> ObservationsExportResponse

    Fetch all observations for a specific trace ID.

    **Parameters:**

    <ParamField path="trace_id" type="string">
      A trace ID.
    </ParamField>

    <ParamField path="page_size" type="int">
      Number of observations per page (1–1000). Default is `100`.
    </ParamField>

    <ParamField path="fetch_all" type="bool">
      When `True`, retrieves observations across all pages automatically. Default is `True`.
    </ParamField>

    <ParamField path="show_progress" type="bool">
      When `True`, logs progress through the logger. Default is `False`.
    </ParamField>

    **Returns:**

    <ResponseField name="ObservationsExportResponse">
      Contains `observations` (list of `Observation` objects), `totalCount`, `page`, and `totalPages`.
    </ResponseField>

    ### export\_trace\_to\_json(self, trace\_id: str, output\_file: Optional\[str] = None, pretty: bool = True, page\_size: int = 50) -> tuple\[ObservationsExportResponse, str]

    Fetch observations for a trace ID and serialize them to JSON. Optionally writes the output to a file.

    **Parameters:**

    <ParamField path="trace_id" type="string">
      A trace ID.
    </ParamField>

    <ParamField path="output_file" type="string (optional)">
      A file path for the JSON output. When `None`, JSON is returned as a string only. Default is `None`.
    </ParamField>

    <ParamField path="pretty" type="bool">
      Indented JSON for readability. Default is `True`.
    </ParamField>

    <ParamField path="page_size" type="int">
      Number of observations per page. Default is `50`.
    </ParamField>

    **Returns:**

    <ResponseField name="tuple">
      * **ObservationsExportResponse:** The observations fetched, for programmatic use.
      * **string:** The JSON output, suitable for display or storage.
    </ResponseField>

    ### search\_traces(self, filters: Optional\[TraceFilters] = None, sort: Optional\[TraceSort] = None, page\_size: int = 100, show\_progress: bool = False) -> TraceSearchResponse

    Search for traces using optional filters and sort options.

    **Parameters:**

    <ParamField path="filters" type="TraceFilters (optional)">
      Search criteria. Supported fields:

      * `start_time` — ISO 8601 string or `datetime`
      * `end_time` — ISO 8601 string or `datetime`
      * `user_ids` — list of user IDs (only the first value is used by the API)
      * `session_ids` — list of session IDs (only the first value is used by the API)

      Default is `None` (returns all traces up to `page_size`).
    </ParamField>

    <ParamField path="sort" type="TraceSort (optional)">
      Sorting options. Use `field="start_time"` and `direction="asc"` or `direction="desc"`. The values `"start_time"` and `"end_time"` both map to `"timestamp"` at the API level.
      Default is `None`.
    </ParamField>

    <ParamField path="page_size" type="int">
      Results per page (1–1000). Default is `100`.
    </ParamField>

    <ParamField path="show_progress" type="bool">
      When `True`, logs progress messages via logger. Default is `False`.
    </ParamField>

    **Returns:**

    <ResponseField name="TraceSearchResponse">
      Contains `traces` (list of `TraceItem`), `traceSummaries`, `totalCount`, and `meta` (pagination).
    </ResponseField>
  </Accordion>
</AccordionGroup>

### Models

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

```python Import example theme={null}
from ibm_watsonx_orchestrate.client.observability.traces.traces_client import (
    TraceFilters,
    TraceSort,
    Observation,
    ObservationsExportResponse,
    TraceItem,
    TraceSearchResponse,
    TraceSummary,
    PaginationMeta,
)
```

<AccordionGroup>
  <Accordion title="Observation">
    Represents a single recorded step within a trace. The API returns this object from `GET /v1/agentops-v3/observations`.

    **Attributes**

    <ParamField path="id" type="string">
      Observation ID.
    </ParamField>

    <ParamField path="traceId" type="string">
      Parent trace ID.
    </ParamField>

    <ParamField path="type" type="string">
      Observation type, for example `GENERATION`.
    </ParamField>

    <ParamField path="name" type="string">
      Observation name.
    </ParamField>

    <ParamField path="startTime" type="string">
      Start time (ISO 8601).
    </ParamField>

    <ParamField path="endTime" type="string | None">
      End time (ISO 8601). May be `None` for in-progress observations.
    </ParamField>

    <ParamField path="model" type="string | None">
      LLM model used, if applicable.
    </ParamField>

    <ParamField path="input" type="dict | list | string | None">
      Input data passed to the operation.
    </ParamField>

    <ParamField path="output" type="dict | list | string | None">
      Output data returned from the operation.
    </ParamField>

    <ParamField path="metadata" type="dict | None">
      Arbitrary metadata.
    </ParamField>

    <ParamField path="usage" type="dict | None">
      Token usage statistics.
    </ParamField>
  </Accordion>

  <Accordion title="ObservationsExportResponse">
    The response object that `fetch_trace_observations()` and `export_trace_to_json()` return.

    **Attributes**

    <ParamField path="observations" type="list[Observation] | None">
      List of observations for the trace.
    </ParamField>

    <ParamField path="totalCount" type="int | None">
      Total number of observations reported by the API. Also accessible as the `total_count` property.
    </ParamField>

    <ParamField path="page" type="int | None">
      Current page number.
    </ParamField>

    <ParamField path="totalPages" type="int | None">
      Total number of pages.
    </ParamField>
  </Accordion>

  <Accordion title="TraceItem">
    Represents a single trace entry. The API returns this object from `GET /v1/agentops-v3/traces`.

    **Attributes**

    <ParamField path="id" type="string">
      Trace ID.
    </ParamField>

    <ParamField path="name" type="string | None">
      Trace name.
    </ParamField>

    <ParamField path="timestamp" type="string">
      Trace start time (ISO 8601).
    </ParamField>

    <ParamField path="sessionId" type="string | None">
      Session ID associated with the trace.
    </ParamField>

    <ParamField path="userId" type="string | None">
      User ID associated with the trace.
    </ParamField>

    <ParamField path="tags" type="list[string] | None">
      Tags attached to the trace.
    </ParamField>

    <ParamField path="latency" type="float | None">
      Trace duration in seconds.
    </ParamField>

    <ParamField path="input" type="dict | list | string | None">
      Input data for the trace.
    </ParamField>

    <ParamField path="output" type="dict | list | string | None">
      Output data for the trace.
    </ParamField>

    <ParamField path="metadata" type="dict | None">
      Arbitrary metadata.
    </ParamField>
  </Accordion>

  <Accordion title="TraceFilters">
    Defines search criteria for trace queries. All fields are optional.

    **Attributes**

    <ParamField path="start_time" type="string | datetime">
      Start of the time range. Accepts an ISO 8601 string or a Python `datetime` object.
    </ParamField>

    <ParamField path="end_time" type="string | datetime">
      End of the time range. Accepts an ISO 8601 string or a Python `datetime` object.
    </ParamField>

    <ParamField path="user_ids" type="list[string]">
      Filter by user ID. Only the first value in the list is sent to the API.
    </ParamField>

    <ParamField path="session_ids" type="list[string]">
      Filter by session ID. Only the first value in the list is sent to the API.
    </ParamField>
  </Accordion>

  <Accordion title="TraceSort">
    Defines sort options for trace search queries.

    **Attributes**

    <ParamField path="field" type="string">
      Field to sort by. Use `"timestamp"`. The values `"start_time"` and `"end_time"` are also accepted and map to `"timestamp"`.
    </ParamField>

    <ParamField path="direction" type="string">
      Sort direction: `"asc"` or `"desc"`.
    </ParamField>
  </Accordion>

  <Accordion title="TraceSummary">
    Provides summary data for a trace. The `TraceSearchResponse.traceSummaries` list contains these objects.

    **Attributes**

    <ParamField path="traceId" type="string">
      Trace ID.
    </ParamField>

    <ParamField path="startTime" type="string">
      Trace start time (ISO 8601).

      <Note>
        **Note:**

        The agentops-v3 API does not return separate start and end times for agent traces. Both `startTime` and `endTime` are populated from the same `timestamp` field returned by the API. Dedicated start/end time values may become available in a future API update.
      </Note>
    </ParamField>

    <ParamField path="endTime" type="string">
      Trace end time (ISO 8601).

      <Note>
        **Note:**

        The agentops-v3 API does not return separate start and end times for agent traces. Both `startTime` and `endTime` are populated from the same `timestamp` field returned by the API. Dedicated start/end time values may become available in a future API update.
      </Note>
    </ParamField>

    <ParamField path="durationMs" type="float">
      Trace duration in milliseconds.
    </ParamField>

    <ParamField path="agentIds" type="list[string] | None">
      Agent IDs extracted from trace metadata.
    </ParamField>

    <ParamField path="agentNames" type="list[string] | None">
      Agent names extracted from trace metadata.
    </ParamField>

    <ParamField path="userIds" type="list[string] | None">
      User IDs associated with the trace.
    </ParamField>

    <ParamField path="sessionIds" type="list[string] | None">
      Session IDs associated with the trace.
    </ParamField>
  </Accordion>

  <Accordion title="TraceSearchResponse">
    The response object that `search_traces()` returns.

    **Attributes**

    <ParamField path="traces" type="list[TraceItem] | None">
      List of trace items from the agentops-v3 API.
    </ParamField>

    <ParamField path="traceSummaries" type="list[TraceSummary]">
      Summary view of each trace, derived from the trace items.
    </ParamField>

    <ParamField path="totalCount" type="int | None">
      Total number of matching traces.
    </ParamField>

    <ParamField path="meta" type="PaginationMeta | None">
      Pagination metadata.
    </ParamField>

    <ParamField path="generatedAt" type="string">
      Time the response object was created.
    </ParamField>

    <ParamField path="originalQuery" type="object">
      The query parameters that were sent to the API.
    </ParamField>
  </Accordion>

  <Accordion title="PaginationMeta">
    Pagination metadata that API responses include.

    **Attributes**

    <ParamField path="page" type="int">
      Current page number.
    </ParamField>

    <ParamField path="limit" type="int">
      Items per page.
    </ParamField>

    <ParamField path="totalItems" type="int">
      Total number of items across all pages.
    </ParamField>

    <ParamField path="totalPages" type="int">
      Total number of pages.
    </ParamField>
  </Accordion>
</AccordionGroup>