strands-token-telemetry 0.1.0

Emit Strands agent token usage as CloudWatch EMF metrics

strands-token-telemetry

Emit Strands Agents token usage as CloudWatch EMF metrics.

Why this library?

Strands Agents has built-in observability via OpenTelemetry traces, and AgentCore adds automatic CloudWatch metrics — but this turnkey telemetry only works when you use BedrockModel and deploy to AgentCore.

If you're already on BedrockModel + AgentCore, you may not need this. Otherwise, this library fills three gaps:

1. Works with any model provider — not locked into BedrockModel. Use Anthropic, LiteLLM, Ollama, or any Strands-compatible model and still get token metrics.

2. Works on any infrastructure — ECS, Lambda, EC2, on-prem, or AgentCore. Anywhere the CloudWatch agent (or a compatible log sink) runs, you get first-class CloudWatch Metrics.

3. First-class CloudWatch Metrics, not trace attributes — Strands' OTEL integration captures token counts as span attributes buried in traces. EMF creates real CloudWatch Metrics you can directly alarm on, dashboard, and apply metric math to (e.g. cost estimation). No OTEL collector setup or ADOT config required — just stdout.

	Strands OTEL traces	AgentCore auto-telemetry	This library
Any model provider	Yes	BedrockModel only	Yes
Any deploy target	Yes	AgentCore only	Yes
CloudWatch Metrics	No (span attributes)	Yes	Yes
CloudWatch Alarms	No	Yes	Yes
OTEL collector needed	Yes	Managed	No
Setup	Configure ADOT/exporter	Deploy to AgentCore	One hook, one line

Install

pip install strands-token-telemetry

strands-agents is a peer dependency — install it separately (or it's likely already in your project).

AI-assisted integration

For developers

Copy and paste this prompt into Claude Code, Cursor, or your preferred AI coding assistant:

Install strands-token-telemetry and add a TokenUsageHook to my Strands agent. Read the integration guide at https://raw.githubusercontent.com/flockcover/strands-token-telemetry/main/docs/llms.txt before writing any code.

For AI coding assistants

If you are an AI coding assistant helping a developer integrate this library, fetch the full API reference and integration patterns from the LLM-optimized docs before generating code:

https://raw.githubusercontent.com/flockcover/strands-token-telemetry/main/docs/llms.txt

Quick start

from strands import Agent
from strands_token_telemetry import TokenUsageHook

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

Every agent invocation prints a JSON line to stdout in CloudWatch EMF format. The CloudWatch agent picks this up and publishes metrics automatically.

Adding session context

A common pattern is tagging metrics with a custom namespace plus user and session identifiers so you can filter and query them in CloudWatch Insights. Pass static dimension values for the model, and extra_properties for fields that should be searchable but not published as metric dimensions:

from strands import Agent
from strands_token_telemetry import TokenUsageHook

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

Model appears as a CloudWatch Metric dimension you can alarm on, while UserId and SessionId stay as top-level properties queryable with CloudWatch Insights (e.g. filter SessionId = "abc-123").

Each invocation emits a single JSON line like this (pretty-printed here for readability):

{
  "_aws": {
    "Timestamp": 1700000000000,
    "CloudWatchMetrics": [
      {
        "Namespace": "AcmeInc/StrandsTokens",
        "Dimensions": [["Model"]],
        "Metrics": [
          { "Name": "inputTokens", "Unit": "Count" },
          { "Name": "outputTokens", "Unit": "Count" },
          { "Name": "totalTokens", "Unit": "Count" },
          { "Name": "cacheReadInputTokens", "Unit": "Count" },
          { "Name": "cacheWriteInputTokens", "Unit": "Count" }
        ]
      }
    ]
  },
  "Model": "us.anthropic.claude-sonnet-4-20250514",
  "UserId": "user-42",
  "SessionId": "abc-123",
  "inputTokens": 1024,
  "outputTokens": 256,
  "totalTokens": 1280,
  "cacheReadInputTokens": 512,
  "cacheWriteInputTokens": 0
}

Configuration

All constructor parameters are keyword-only.

Parameter	Type	Default	Description
namespace	str	"Strands/AgentTokenUsage"	CloudWatch metrics namespace
dimensions	list[list[str]]	[["Model"]]	Dimension key sets
dimension_values	dict[str, str]	{}	Static dimension key/value pairs
dimension_resolver	Callable	None	Receives AfterInvocationEvent, returns dynamic dimension values
extra_properties	dict[str, Any]	None	Extra top-level properties (searchable in CloudWatch Insights)
emitter	Callable	default_emitter	Function that receives the payload dict

Dynamic dimensions

Use dimension_resolver when a dimension value isn't known until the agent runs — for example, the model name returned by the provider, or an agent identifier pulled from the event. Static values like environment or service name can go in dimension_values; the resolver handles everything that changes per invocation.

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

A more advanced example — splitting metrics by both model and a per-request agent name:

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

agent = Agent(hooks=[
    TokenUsageHook(
        dimensions=[["Model", "AgentName"]],
        dimension_resolver=resolve_dims,
    )
])

Custom emitter

By default the hook prints compact JSON to stdout, which the CloudWatch agent picks up. Replace the emitter when you need the payload to go somewhere else — for example, sending metrics to a non-CloudWatch backend or routing through your application's structured logging pipeline.

import json
import logging

logger = logging.getLogger("token_metrics")

def log_emitter(payload):
    logger.info(json.dumps(payload))

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

You can also forward to an external service:

import json
import 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)

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

Local development

When you run an agent locally the default emitter prints one compact JSON line to stdout on every invocation. For example:

{"_aws":{"Timestamp":1700000000000,"CloudWatchMetrics":[...]},"inputTokens":42,...}

This is normal — it is CloudWatch Embedded Metric Format (EMF) output that the CloudWatch agent would consume in production. Locally there is no CloudWatch agent, so the lines simply appear in your console.

Suppressing output

Pass a no-op emitter to silence the JSON lines entirely:

from strands_token_telemetry import TokenUsageHook

hook = TokenUsageHook(emitter=lambda payload: None)

Human-readable output

Pretty-print the payload so you can inspect it during development:

import json
from strands_token_telemetry import TokenUsageHook

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

Logging instead of stdout

Route output through Python's logging module so it respects your existing log configuration:

import json
import logging
from strands_token_telemetry import TokenUsageHook

log = logging.getLogger("token_telemetry")

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

Development

pip install -e ".[dev]"
pytest -v