Usage

RequestUsage

Usage details for a single API request.

Source code in src/agents/usage.py

@dataclass
class RequestUsage:
    """Usage details for a single API request."""

    input_tokens: int
    """Input tokens for this individual request."""

    output_tokens: int
    """Output tokens for this individual request."""

    total_tokens: int
    """Total tokens (input + output) for this individual request."""

    input_tokens_details: InputTokensDetails
    """Details about the input tokens for this individual request."""

    output_tokens_details: OutputTokensDetails
    """Details about the output tokens for this individual request."""

input_tokens instance-attribute

input_tokens: int

Input tokens for this individual request.

output_tokens instance-attribute

output_tokens: int

Output tokens for this individual request.

total_tokens instance-attribute

total_tokens: int

Total tokens (input + output) for this individual request.

input_tokens_details instance-attribute

input_tokens_details: InputTokensDetails

Details about the input tokens for this individual request.

output_tokens_details instance-attribute

output_tokens_details: OutputTokensDetails

Details about the output tokens for this individual request.

Usage

Source code in src/agents/usage.py

@dataclass
class Usage:
    requests: int = 0
    """Total requests made to the LLM API."""

    input_tokens: int = 0
    """Total input tokens sent, across all requests."""

    input_tokens_details: Annotated[
        InputTokensDetails, BeforeValidator(_normalize_input_tokens_details)
    ] = field(default_factory=_make_input_tokens_details)
    """Details about the input tokens, matching responses API usage details."""
    output_tokens: int = 0
    """Total output tokens received, across all requests."""

    output_tokens_details: Annotated[
        OutputTokensDetails, BeforeValidator(_normalize_output_tokens_details)
    ] = field(default_factory=lambda: OutputTokensDetails(reasoning_tokens=0))
    """Details about the output tokens, matching responses API usage details."""

    total_tokens: int = 0
    """Total tokens sent and received, across all requests."""

    request_usage_entries: list[RequestUsage] = field(default_factory=list)
    """List of RequestUsage entries for accurate per-request cost calculation.

    Each call to `add()` automatically creates an entry in this list if the added usage
    represents a new request (i.e., has non-zero tokens).

    Example:
        For a run that makes 3 API calls with 100K, 150K, and 80K input tokens each,
        the aggregated `input_tokens` would be 330K, but `request_usage_entries` would
        preserve the [100K, 150K, 80K] breakdown, which could be helpful for detailed
        cost calculation or context window management.
    """

    def __post_init__(self) -> None:
        # Some providers don't populate optional token detail fields
        # (cached_tokens, cache_write_tokens, reasoning_tokens), and the OpenAI SDK's generated
        # code can bypass Pydantic validation (e.g., via model_construct),
        # allowing None values. We normalize these to 0 to prevent TypeErrors.
        input_details_none = self.input_tokens_details is None
        input_cached_none = (
            not input_details_none and self.input_tokens_details.cached_tokens is None
        )
        input_cache_write_none = (
            not input_details_none
            and getattr(self.input_tokens_details, "cache_write_tokens", 0) is None
        )
        if input_details_none or input_cached_none or input_cache_write_none:
            self.input_tokens_details = _make_input_tokens_details(
                cached_tokens=_cached_tokens(self.input_tokens_details),
                cache_write_tokens=_cache_write_tokens(self.input_tokens_details),
            )

        output_details_none = self.output_tokens_details is None
        output_reasoning_none = (
            not output_details_none and self.output_tokens_details.reasoning_tokens is None
        )
        if output_details_none or output_reasoning_none:
            self.output_tokens_details = OutputTokensDetails(reasoning_tokens=0)

    def add(self, other: Usage) -> None:
        """Add another Usage object to this one, aggregating all fields.

        This method automatically preserves request_usage_entries.

        Args:
            other: The Usage object to add to this one.
        """
        self.requests += other.requests if other.requests else 0
        self.input_tokens += other.input_tokens if other.input_tokens else 0
        self.output_tokens += other.output_tokens if other.output_tokens else 0
        self.total_tokens += other.total_tokens if other.total_tokens else 0

        # Null guards for nested token details (other may bypass validation via model_construct)
        other_cached = _cached_tokens(other.input_tokens_details)
        other_cache_write = _cache_write_tokens(other.input_tokens_details)
        other_reasoning = (
            other.output_tokens_details.reasoning_tokens
            if other.output_tokens_details and other.output_tokens_details.reasoning_tokens
            else 0
        )
        self_cached = _cached_tokens(self.input_tokens_details)
        self_cache_write = _cache_write_tokens(self.input_tokens_details)
        self_reasoning = (
            self.output_tokens_details.reasoning_tokens
            if self.output_tokens_details and self.output_tokens_details.reasoning_tokens
            else 0
        )

        self.input_tokens_details = _make_input_tokens_details(
            cached_tokens=self_cached + other_cached,
            cache_write_tokens=self_cache_write + other_cache_write,
        )

        self.output_tokens_details = OutputTokensDetails(
            reasoning_tokens=self_reasoning + other_reasoning
        )

        # Automatically preserve request_usage_entries.
        # If the other Usage already has individual request breakdowns, merge them
        # (this preserves nested token details that would otherwise be discarded
        # when synthesizing an entry from only the top-level fields).
        if other.request_usage_entries:
            self.request_usage_entries.extend(other.request_usage_entries)
        elif other.requests == 1 and other.total_tokens > 0:
            # Otherwise, if the other Usage represents a single request with tokens, record it.
            input_details = other.input_tokens_details or _make_input_tokens_details()
            output_details = other.output_tokens_details or OutputTokensDetails(reasoning_tokens=0)
            request_usage = RequestUsage(
                input_tokens=other.input_tokens,
                output_tokens=other.output_tokens,
                total_tokens=other.total_tokens,
                input_tokens_details=input_details,
                output_tokens_details=output_details,
            )
            self.request_usage_entries.append(request_usage)

requests class-attribute instance-attribute

requests: int = 0

Total requests made to the LLM API.

input_tokens class-attribute instance-attribute

input_tokens: int = 0

Total input tokens sent, across all requests.

input_tokens_details class-attribute instance-attribute

input_tokens_details: Annotated[
    InputTokensDetails,
    BeforeValidator(_normalize_input_tokens_details),
] = field(default_factory=_make_input_tokens_details)

Details about the input tokens, matching responses API usage details.

output_tokens class-attribute instance-attribute

output_tokens: int = 0

Total output tokens received, across all requests.

output_tokens_details class-attribute instance-attribute

output_tokens_details: Annotated[
    OutputTokensDetails,
    BeforeValidator(_normalize_output_tokens_details),
] = field(
    default_factory=lambda: OutputTokensDetails(
        reasoning_tokens=0
    )
)

Details about the output tokens, matching responses API usage details.

total_tokens class-attribute instance-attribute

total_tokens: int = 0

Total tokens sent and received, across all requests.

request_usage_entries class-attribute instance-attribute

request_usage_entries: list[RequestUsage] = field(
    default_factory=list
)

List of RequestUsage entries for accurate per-request cost calculation.

Each call to add() automatically creates an entry in this list if the added usage represents a new request (i.e., has non-zero tokens).

Example

For a run that makes 3 API calls with 100K, 150K, and 80K input tokens each, the aggregated input_tokens would be 330K, but request_usage_entries would preserve the [100K, 150K, 80K] breakdown, which could be helpful for detailed cost calculation or context window management.

add

add(other: Usage) -> None

Add another Usage object to this one, aggregating all fields.

This method automatically preserves request_usage_entries.

Parameters:

Name	Type	Description	Default
other	Usage	The Usage object to add to this one.	required

Source code in src/agents/usage.py

def add(self, other: Usage) -> None:
    """Add another Usage object to this one, aggregating all fields.

    This method automatically preserves request_usage_entries.

    Args:
        other: The Usage object to add to this one.
    """
    self.requests += other.requests if other.requests else 0
    self.input_tokens += other.input_tokens if other.input_tokens else 0
    self.output_tokens += other.output_tokens if other.output_tokens else 0
    self.total_tokens += other.total_tokens if other.total_tokens else 0

    # Null guards for nested token details (other may bypass validation via model_construct)
    other_cached = _cached_tokens(other.input_tokens_details)
    other_cache_write = _cache_write_tokens(other.input_tokens_details)
    other_reasoning = (
        other.output_tokens_details.reasoning_tokens
        if other.output_tokens_details and other.output_tokens_details.reasoning_tokens
        else 0
    )
    self_cached = _cached_tokens(self.input_tokens_details)
    self_cache_write = _cache_write_tokens(self.input_tokens_details)
    self_reasoning = (
        self.output_tokens_details.reasoning_tokens
        if self.output_tokens_details and self.output_tokens_details.reasoning_tokens
        else 0
    )

    self.input_tokens_details = _make_input_tokens_details(
        cached_tokens=self_cached + other_cached,
        cache_write_tokens=self_cache_write + other_cache_write,
    )

    self.output_tokens_details = OutputTokensDetails(
        reasoning_tokens=self_reasoning + other_reasoning
    )

    # Automatically preserve request_usage_entries.
    # If the other Usage already has individual request breakdowns, merge them
    # (this preserves nested token details that would otherwise be discarded
    # when synthesizing an entry from only the top-level fields).
    if other.request_usage_entries:
        self.request_usage_entries.extend(other.request_usage_entries)
    elif other.requests == 1 and other.total_tokens > 0:
        # Otherwise, if the other Usage represents a single request with tokens, record it.
        input_details = other.input_tokens_details or _make_input_tokens_details()
        output_details = other.output_tokens_details or OutputTokensDetails(reasoning_tokens=0)
        request_usage = RequestUsage(
            input_tokens=other.input_tokens,
            output_tokens=other.output_tokens,
            total_tokens=other.total_tokens,
            input_tokens_details=input_details,
            output_tokens_details=output_details,
        )
        self.request_usage_entries.append(request_usage)

deserialize_usage

deserialize_usage(usage_data: Mapping[str, Any]) -> Usage

Rebuild a Usage object from serialized JSON data.

Source code in src/agents/usage.py

def deserialize_usage(usage_data: Mapping[str, Any]) -> Usage:
    """Rebuild a Usage object from serialized JSON data."""
    input_tokens_details_raw = usage_data.get("input_tokens_details")
    output_tokens_details_raw = usage_data.get("output_tokens_details")
    input_details = _coerce_input_token_details(input_tokens_details_raw)
    output_details = _coerce_token_details(
        TypeAdapter(OutputTokensDetails),
        output_tokens_details_raw or {"reasoning_tokens": 0},
        OutputTokensDetails(reasoning_tokens=0),
    )

    request_entries: list[RequestUsage] = []
    request_entries_raw = usage_data.get("request_usage_entries") or []
    for entry in request_entries_raw:
        request_entries.append(
            RequestUsage(
                input_tokens=entry.get("input_tokens", 0),
                output_tokens=entry.get("output_tokens", 0),
                total_tokens=entry.get("total_tokens", 0),
                input_tokens_details=_coerce_input_token_details(entry.get("input_tokens_details")),
                output_tokens_details=_coerce_token_details(
                    TypeAdapter(OutputTokensDetails),
                    entry.get("output_tokens_details") or {"reasoning_tokens": 0},
                    OutputTokensDetails(reasoning_tokens=0),
                ),
            )
        )

    return Usage(
        requests=usage_data.get("requests", 0),
        input_tokens=usage_data.get("input_tokens", 0),
        output_tokens=usage_data.get("output_tokens", 0),
        total_tokens=usage_data.get("total_tokens", 0),
        input_tokens_details=input_details,
        output_tokens_details=output_details,
        request_usage_entries=request_entries,
    )

serialize_usage

serialize_usage(usage: Usage) -> dict[str, Any]

Serialize a Usage object into a JSON-friendly dictionary.

Source code in src/agents/usage.py

def serialize_usage(usage: Usage) -> dict[str, Any]:
    """Serialize a Usage object into a JSON-friendly dictionary."""
    input_details = _serialize_input_tokens_details(usage.input_tokens_details)
    output_details = _serialize_usage_details(usage.output_tokens_details, {"reasoning_tokens": 0})

    def _serialize_request_entry(entry: RequestUsage) -> dict[str, Any]:
        return {
            "input_tokens": entry.input_tokens,
            "output_tokens": entry.output_tokens,
            "total_tokens": entry.total_tokens,
            "input_tokens_details": _serialize_input_tokens_details(entry.input_tokens_details),
            "output_tokens_details": _serialize_usage_details(
                entry.output_tokens_details, {"reasoning_tokens": 0}
            ),
        }

    return {
        "requests": usage.requests,
        "input_tokens": usage.input_tokens,
        "input_tokens_details": [input_details],
        "output_tokens": usage.output_tokens,
        "output_tokens_details": [output_details],
        "total_tokens": usage.total_tokens,
        "request_usage_entries": [
            _serialize_request_entry(entry) for entry in usage.request_usage_entries
        ],
    }

model_usage_to_span_usage

model_usage_to_span_usage(usage: Usage) -> dict[str, Any]

Serialize full per-model-call usage for tracing span data.

Source code in src/agents/usage.py

def model_usage_to_span_usage(usage: Usage) -> dict[str, Any]:
    """Serialize full per-model-call usage for tracing span data."""
    return {
        "requests": usage.requests,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "total_tokens": usage.total_tokens,
        "input_tokens_details": _serialize_input_tokens_details(usage.input_tokens_details),
        "output_tokens_details": _serialize_usage_details(
            usage.output_tokens_details,
            {"reasoning_tokens": 0},
        ),
    }

total_usage_to_span_metadata

total_usage_to_span_metadata(
    usage: Usage,
) -> dict[str, int]

Serialize aggregate task/run usage for tracing span metadata.

Source code in src/agents/usage.py

def total_usage_to_span_metadata(usage: Usage) -> dict[str, int]:
    """Serialize aggregate task/run usage for tracing span metadata."""
    return {
        "requests": usage.requests,
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "total_tokens": usage.total_tokens,
        "cached_input_tokens": _cached_input_tokens(usage),
        "cache_write_input_tokens": _cache_write_input_tokens(usage),
    }

turn_usage_to_span_data

turn_usage_to_span_data(usage: Usage) -> dict[str, int]

Serialize aggregate per-turn usage for custom turn span data.

Source code in src/agents/usage.py

def turn_usage_to_span_data(usage: Usage) -> dict[str, int]:
    """Serialize aggregate per-turn usage for custom turn span data."""
    return {
        "input_tokens": usage.input_tokens,
        "output_tokens": usage.output_tokens,
        "cached_input_tokens": _cached_input_tokens(usage),
        "cache_write_input_tokens": _cache_write_input_tokens(usage),
    }

task_usage_to_span_data

task_usage_to_span_data(usage: Usage) -> dict[str, int]

Serialize aggregate per-task usage for custom task span data.

Source code in src/agents/usage.py

def task_usage_to_span_data(usage: Usage) -> dict[str, int]:
    """Serialize aggregate per-task usage for custom task span data."""
    return {
        **turn_usage_to_span_data(usage),
        "requests": usage.requests,
        "total_tokens": usage.total_tokens,
    }