Using the Economics Engine API

This guide describes how to use the unomiq-sdk Python package to query the Unomiq Economic Engine API — the API that lets you search traces, retrieve aggregated metrics, list traces, and inspect individual spans.

Prerequisites

API Credentials

Create API credentials from the Unomiq Dashboard. The credentials must have the read:traces permission. This will give you an API key (Client ID) and secret (Client Secret).

Application ID

You need the Application ID (app_id) for the application whose traces you want to query. You can find this in the Unomiq Dashboard under your application settings.

Install the SDK

pip install unomiq-sdk

Configuration

Set the following environment variables for your application:

Variable	Description
`UNOMIQ_CLIENT_ID`	API Key from the Unomiq Dashboard
`UNOMIQ_CLIENT_SECRET`	API Secret from the Unomiq Dashboard

The SDK uses these credentials internally to acquire and refresh OAuth2 tokens when making requests to the Engine API. Alternatively, you can pass credentials directly to the constructor:

from unomiq import UnomiqEconomicsEngine

engine = UnomiqEconomicsEngine(
    app_id="my-app-id",
    client_id="your-client-id",
    client_secret="your-client-secret",
)

Constructor Parameters

Parameter	Type	Default	Description
`app_id`	`str`	required	Application ID (sent as `X-Unomiq-App-Id` header)
`client_id`	`str \| None`	`None`	OAuth2 client ID (falls back to `UNOMIQ_CLIENT_ID` env var)
`client_secret`	`str \| None`	`None`	OAuth2 client secret (falls back to `UNOMIQ_CLIENT_SECRET` env var)
`api_endpoint`	`str`	`https://engine-api.unomiq.com`	Engine API endpoint
`token_url`	`str`	`https://oauth-api.unomiq.com/token`	OAuth2 token endpoint
`scopes`	`str`	`read:traces`	OAuth2 scopes
`audience`	`str`	`https://engine-api.unomiq.com`	OAuth2 audience

Initializing the Client

Using environment variables (recommended):

from unomiq import UnomiqEconomicsEngine

engine = UnomiqEconomicsEngine(app_id="my-app-id")

Remember to call engine.close() when you are done to release resources.

Querying Spans

The engine/query endpoint lets you search for individual spans across traces using filter expressions.

from datetime import datetime
from unomiq import UnomiqEconomicsEngine, Condition, QueryOperator

engine = UnomiqEconomicsEngine(app_id="my-app-id")

results = engine.query(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
    filter=Condition(field="service_name", op=QueryOperator.EQ, value="my-service"),
)

for trace in results.data:
    print(f"{trace.trace_id} — {trace.span_name} ({trace.duration_ms}ms)")

engine.close()

The response includes paginated span data along with the total count of matching results.

Filter Expressions

The SDK provides composable filter primitives — Condition, AndGroup, and OrGroup — that map directly to the Engine API filter schema. These can be used with engine.query(), engine.get_traces(), and engine.get_metrics().

Simple Condition

A Condition matches a single field against a value using a comparison operator:

from unomiq import Condition, QueryOperator

filter = Condition(field="service_name", op=QueryOperator.EQ, value="my-service")

Supported Operators

Operator	Description
`QueryOperator.EQ`	Equal to
`QueryOperator.NEQ`	Not equal to
`QueryOperator.CONTAINS`	Contains substring
`QueryOperator.STARTS_WITH`	Starts with
`QueryOperator.ENDS_WITH`	Ends with
`QueryOperator.GT`	Greater than
`QueryOperator.LT`	Less than
`QueryOperator.GTE`	Greater than or equal to
`QueryOperator.LTE`	Less than or equal to

Filterable Fields

Field	Description
`trace_id`	Trace identifier
`span_id`	Span identifier
`span_name`	Span name
`span_kind`	Span kind (e.g., `SERVER`, `CLIENT`)
`service_name`	Service name
`unit`	Unit (equivalent to `span_attributes.unomiq.unit`)
`parent_unit`	Parent unit (equivalent to `span_attributes.unomiq.parent_unit`)
`duration_ms`	Span duration in milliseconds
`span_attributes.*`	Any span attribute (e.g., `span_attributes.http.method`)
`resource_attributes.*`	Any resource attribute
`cost`	Matched cost
`billing.resource_type`	Billing resource type (e.g., `db_job`, `api_request`, `llm_generate_content`)
`billing.resource_name`	Billing resource name
`billing.service_id`	Billing service ID
`billing.service_name`	Billing service name
`billing.sku_id`	Billing SKU ID
`billing.sku_name`	Billing SKU name
`billing.region`	Billing region
`usage.amount`	Usage amount
`usage.unit`	Usage unit

Combining Conditions with AND / OR

Use AndGroup and OrGroup to build complex filter expressions:

from unomiq import AndGroup, OrGroup, Condition, QueryOperator

filter = AndGroup(and_=[
    Condition(field="span_attributes.http.status_code", op=QueryOperator.GTE, value="200"),
    Condition(field="span_attributes.http.status_code", op=QueryOperator.LTE, value="299"),
    OrGroup(or_=[
        Condition(field="span_kind", op=QueryOperator.EQ, value="SERVER"),
        Condition(field="span_kind", op=QueryOperator.EQ, value="CLIENT"),
    ]),
])

results = engine.query(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
    filter=filter,
)

Groups can be nested arbitrarily to express any boolean logic.

Listing Traces

The /traces endpoint returns a paginated list of traces with aggregated information (duration, span count, total cost).

traces = engine.get_traces(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
)

for t in traces.data:
    print(f"{t.trace_id} — {t.span_name} ({t.spans_count} spans, ${t.total_cost})")

By default, filters apply only to root spans. Set search_child_spans=True to apply filters across all spans in the trace:

traces = engine.get_traces(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
    search_child_spans=True,
    filter=Condition(field="service_name", op=QueryOperator.EQ, value="my-service"),
)

Retrieving a Single Trace

The /traces/{trace_id} endpoint returns all spans for a specific trace:

spans = engine.get_trace(
    trace_id="4bf92f3577b34da6a3ce929d0e0e4736",
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
)

for span in spans:
    print(f"  {span.span_id} {span.span_name} ({span.span_kind})")

Getting Metrics

The /engine/metrics endpoint returns aggregated statistical distributions for trace duration, cost, and spans per trace.

metrics = engine.get_metrics(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
)

print(f"Traces: {metrics.trace_count}")
print(f"Duration p50: {metrics.duration.p50}ms, p99: {metrics.duration.p99}ms")
print(f"Cost mean: {metrics.cost.mean}")

You can pass the same filter expressions to narrow down the metrics:

metrics = engine.get_metrics(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
    filter=Condition(field="service_name", op=QueryOperator.EQ, value="my-service"),
)

The response includes distribution statistics (mean, median, p50, p90, p95, p99, min, max, stddev) for each metric.

Live Traces

The /traces/live endpoint returns the most recent spans from the live data stream:

live = engine.get_live_traces(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
)

for span in live.data:
    print(f"{span.trace_id}/{span.span_id} — {span.span_name}")

Cleanup

Always close the engine client when you are done to release the underlying HTTP connection and OAuth2 token manager:

engine.close()

Full Example

Here is a complete example that initializes the client, queries spans, retrieves metrics, and lists traces:

from datetime import datetime
from unomiq import (
    UnomiqEconomicsEngine,
    Condition,
    AndGroup,
    OrGroup,
    QueryOperator,
)

engine = UnomiqEconomicsEngine(app_id="my-app-id")

# Query spans for a specific service
results = engine.query(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
    filter=AndGroup(and_=[
        Condition(field="service_name", op=QueryOperator.EQ, value="my-service"),
        Condition(field="duration_ms", op=QueryOperator.GTE, value="100"),
    ]),
)
print(f"Found {results.total} matching spans")

for span in results.data:
    print(f"  {span.trace_id} — {span.span_name} ({span.duration_ms}ms)")

# Get aggregated metrics
metrics = engine.get_metrics(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
)
print(f"\nTotal traces: {metrics.trace_count}")
print(f"Duration p50: {metrics.duration.p50}ms, p99: {metrics.duration.p99}ms")
print(f"Cost mean: ${metrics.cost.mean:.4f}")

# List traces with cost information
traces = engine.get_traces(
    start_time=datetime(2026, 1, 15),
    end_time=datetime(2026, 1, 16),
)
for t in traces.data:
    print(f"  {t.trace_id} — {t.span_name} ({t.spans_count} spans, ${t.total_cost})")

# Inspect a specific trace
if traces.data:
    spans = engine.get_trace(
        trace_id=traces.data[0].trace_id,
        start_time=datetime(2026, 1, 15),
        end_time=datetime(2026, 1, 16),
    )
    print(f"\nSpans in trace {traces.data[0].trace_id}:")
    for span in spans:
        print(f"  {span.span_id} {span.span_name} ({span.span_kind})")

engine.close()

Getting started

Sending Data

Querying Data

Cost Attribution

Changelog

Using the Economics Engine API

Prerequisites

API Credentials

Application ID

Install the SDK

Configuration

Constructor Parameters

Initializing the Client

Querying Spans

Filter Expressions

Simple Condition

Supported Operators

Filterable Fields

Combining Conditions with AND / OR

Listing Traces

Retrieving a Single Trace

Getting Metrics

Live Traces

Cleanup

Full Example

Getting started

Sending Data

Querying Data

Cost Attribution

Changelog

​Prerequisites

​API Credentials

​Application ID

​Install the SDK

​Configuration

​Constructor Parameters

​Initializing the Client

​Querying Spans

​Filter Expressions

​Simple Condition

​Supported Operators

​Filterable Fields

​Combining Conditions with AND / OR

​Listing Traces

​Retrieving a Single Trace

​Getting Metrics

​Live Traces

​Cleanup

​Full Example

Prerequisites

API Credentials

Application ID

Install the SDK

Configuration

Constructor Parameters

Initializing the Client

Querying Spans

Filter Expressions

Simple Condition

Supported Operators

Filterable Fields

Combining Conditions with AND / OR

Listing Traces

Retrieving a Single Trace

Getting Metrics

Live Traces

Cleanup

Full Example