strands-token-telemetry 0.1.0
Emit Strands agent token usage as CloudWatch EMF metrics
Requires Python: >=3.10

# strands-token-telemetry

Emit [Strands Agents](https://github.com/strands-agents/sdk-python) token usage as [CloudWatch EMF](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format.html) 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

```bash
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

```python
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](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format.html) 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:

```python
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):

```json
{
  "_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.

```python
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:

```python
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.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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

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