fix(event_loop): handle MetadataEvents without optional usage and metrics

[dbschmigelski]

Description

This PR resolves issue #1158 where MetadataEvent falsely claimed that metrics and usage fields were optional through type hints, but the runtime implementation required them to be present, causing KeyError exceptions.

The MetadataEvent TypedDict was defined with total=False, making all fields appear optional in type annotations:

class MetadataEvent(TypedDict, total=False):
    """Event containing metadata about the streaming response.

    Attributes:
        metrics: Performance metrics related to the model invocation.
        trace: Trace information for debugging and monitoring.
        usage: Resource usage information for the model invocation.
    """

    metrics: Metrics
    trace: Optional[Trace]
    usage: Usage

However, the extract_usage_metrics function assumed both usage and metrics fields were always present, causing runtime failures when custom model implementations omitted these fields.

Solution

We modified the extract_usage_metrics function to handle optional fields gracefully by providing sensible defaults for the required fields in both Usage and Metrics types:

class Usage(TypedDict, total=False):
    """Token usage information for model interactions."""
    inputTokens: Required[int]
    outputTokens: Required[int]
    totalTokens: Required[int]
    cacheReadInputTokens: int
    cacheWriteInputTokens: int

class Metrics(TypedDict, total=False):
    """Performance metrics for model interactions."""
    latencyMs: Required[int]
    timeToFirstByteMs: int

Design Decision

We chose to make the runtime behavior match the type hints rather than changing the type definitions because this approach maintains backward compatibility while enabling the flexibility that custom model implementations need. This allows models that don't have access to certain metrics (like latency information) to omit those fields without causing runtime errors.

In general, the defaulting is not ideal compared to None. However, the Usage and Metrics being present is assumed throughout the SDK so we cannot make that backwards incompatible change.

Related Issues

#1158

Documentation PR

N/A

Type of Change

Bug fix

Testing

How have you tested the change? Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

• I ran hatch run prepare

Also ran

"""Integration test for MetadataEvent with optional metrics field."""

import asyncio
from typing import AsyncGenerator, Optional

from strands import Agent
from strands.models import Model
from strands.types.content import Messages
from strands.types.streaming import MessageStartEvent, MessageStopEvent, MetadataEvent, StreamEvent
from strands.types.tools import ToolChoice, ToolSpec


class MinimalModelWithoutMetrics(Model):
    """Minimal model implementation that omits metrics in MetadataEvent."""

    async def stream(
        self,
        messages: Messages,
        tool_specs: Optional[list[ToolSpec]] = None,
        system_prompt: Optional[str] = None,
        *,
        tool_choice: ToolChoice | None = None,
        **kwargs
    ) -> AsyncGenerator[StreamEvent, None]:
        yield StreamEvent(messageStart=MessageStartEvent(role="assistant"))
        yield StreamEvent(contentBlockStart={"contentBlockIndex": 0, "start": {}})
        yield StreamEvent(contentBlockDelta={"delta": {"text": "Hi"}, "contentBlockIndex": 0})
        yield StreamEvent(contentBlockStop={"contentBlockIndex": 0})
        yield StreamEvent(messageStop=MessageStopEvent(stopReason="end_turn"))
        # MetadataEvent without metrics field - this should not raise KeyError
        yield StreamEvent(metadata=MetadataEvent(usage={"inputTokens": 5, "outputTokens": 2, "totalTokens": 7}))

    async def structured_output(self, *args, **kwargs):
        raise NotImplementedError()

    def get_config(self) -> dict:
        return {}

    def update_config(self, **kwargs) -> None:
        pass


class MinimalModelWithoutUsage(Model):
    """Minimal model implementation that omits usage in MetadataEvent."""

    async def stream(
        self,
        messages: Messages,
        tool_specs: Optional[list[ToolSpec]] = None,
        system_prompt: Optional[str] = None,
        *,
        tool_choice: ToolChoice | None = None,
        **kwargs
    ) -> AsyncGenerator[StreamEvent, None]:
        yield StreamEvent(messageStart=MessageStartEvent(role="assistant"))
        yield StreamEvent(contentBlockStart={"contentBlockIndex": 0, "start": {}})
        yield StreamEvent(contentBlockDelta={"delta": {"text": "Hello"}, "contentBlockIndex": 0})
        yield StreamEvent(contentBlockStop={"contentBlockIndex": 0})
        yield StreamEvent(messageStop=MessageStopEvent(stopReason="end_turn"))
        # MetadataEvent without usage field - this should not raise KeyError
        yield StreamEvent(metadata=MetadataEvent(metrics={"latencyMs": 100}))

    async def structured_output(self, *args, **kwargs):
        raise NotImplementedError()

    def get_config(self) -> dict:
        return {}

    def update_config(self, **kwargs) -> None:
        pass


def test_metadata_event_without_metrics():
    """Test that MetadataEvent works without metrics field."""
    result = asyncio.run(Agent(model=MinimalModelWithoutMetrics()).invoke_async("test"))
    assert result.message['content'][0]['text'] == "Hi"


def test_metadata_event_without_usage():
    """Test that MetadataEvent works without usage field."""
    result = asyncio.run(Agent(model=MinimalModelWithoutUsage()).invoke_async("test"))
    assert result.message['content'][0]['text'] == "Hell

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!