# strands-token-telemetry

> Emit Strands Agents token usage as CloudWatch Embedded Metric Format (EMF) metrics.

## Install

pip install strands-token-telemetry

strands-agents is a peer dependency (>= 0.1.0).

## Public API

The package exports three names from strands_token_telemetry:

- TokenUsageHook — the main hook class (HookProvider)
- build_emf_payload — pure function to build an EMF dict
- default_emitter — prints compact JSON to stdout

## TokenUsageHook

A Strands HookProvider. Pass it in the hooks list when creating an Agent.

Constructor (all parameters are keyword-only):

  namespace: str = "Strands/AgentTokenUsage"
      CloudWatch metrics namespace.

  dimensions: list[list[str]] = [["Model"]]
      Dimension key sets for CloudWatch metrics.

  dimension_values: dict[str, str] = {}
      Static dimension key/value pairs. Keys must appear in dimensions.

  dimension_resolver: Callable[[AfterInvocationEvent], dict[str, str]] | None = None
      Called on each invocation with the event. Returns dimension key/value
      pairs that are merged with dimension_values (resolver wins on conflict).
      Use this when a value isn't known until runtime (e.g. model name from
      the provider response).

  extra_properties: dict[str, Any] | None = None
      Additional top-level properties in the EMF payload. These are NOT
      published as CloudWatch Metric dimensions but ARE queryable in
      CloudWatch Logs Insights. Use for user IDs, session IDs, trace IDs, etc.

  emitter: Callable[[dict[str, Any]], None] = default_emitter
      Function that receives the final payload dict. Override to route output
      to logging, a webhook, or /dev/null.

## Minimal usage

from strands import Agent
from strands_token_telemetry import TokenUsageHook

agent = Agent(hooks=[TokenUsageHook()])

## Common patterns

### Custom namespace with session context

hook = TokenUsageHook(
    namespace="AcmeInc/StrandsTokens",
    dimension_values={"Model": model_id},
    extra_properties={"UserId": user_id, "SessionId": session_id},
)
agent = Agent(hooks=[hook])

### Dynamic dimensions resolved at runtime

def resolve_dims(event):
    model = getattr(event.result, "model", "unknown") if event.result else "unknown"
    return {"Model": model}

agent = Agent(hooks=[
    TokenUsageHook(
        dimensions=[["Model", "Environment"]],
        dimension_values={"Environment": "prod"},
        dimension_resolver=resolve_dims,
    )
])

### Suppress stdout output (local development)

hook = TokenUsageHook(emitter=lambda payload: None)

### Pretty-print for debugging

import json
hook = TokenUsageHook(emitter=lambda p: print(json.dumps(p, indent=2)))

### Route to Python logging

import json, logging
log = logging.getLogger("token_telemetry")
hook = TokenUsageHook(emitter=lambda p: log.debug("%s", json.dumps(p)))

### Forward to an external service

import json, urllib.request

def webhook_emitter(payload):
    req = urllib.request.Request(
        "https://metrics.example.com/ingest",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
    )
    urllib.request.urlopen(req)

hook = TokenUsageHook(emitter=webhook_emitter)

## build_emf_payload

Pure function — useful if you want to build the EMF dict without the hook.

build_emf_payload(
    usage: dict[str, int],        # token counters (each becomes a metric)
    namespace: str,
    dimensions: list[list[str]],
    dimension_values: dict[str, str],
    extra_properties: dict[str, Any] | None = None,
    timestamp_ms: int | None = None,  # defaults to now
) -> dict[str, Any]

## Metrics emitted

The hook emits every key present in the SDK's usage dict as a Count metric.
Typical keys: inputTokens, outputTokens, totalTokens, cacheReadInputTokens,
cacheWriteInputTokens.

## Key concepts

- dimension_values vs extra_properties: dimension_values become CloudWatch
  Metric dimensions (you can alarm/filter on them). extra_properties are
  top-level JSON fields queryable in CloudWatch Logs Insights but not
  published as metric dimensions.

- dimension_resolver vs dimension_values: use dimension_values for values
  known at hook construction time (environment, service name). Use
  dimension_resolver for values only available at invocation time (model
  name from the response, agent name).

- emitter: the default prints compact JSON to stdout for the CloudWatch
  agent. Override it to send elsewhere or suppress output.

## Source

https://github.com/flockcover/strands-token-telemetry