Token usage becomes an infrastructure problem as soon as someone asks which application or user caused the bill. A provider total tells you how much you consumed. It rarely gives your application the attribution it needs to make a useful decision.

A model bill with only aggregate totals is like a shared utility bill in a building with no meters. The total is real but deciding where to reduce consumption is guesswork.

My default is to capture usage at the model-call boundary, while the application still has the context needed to label it. This walkthrough sends the usage reported by a completed Bedrock call to Amazon Simple Queue Service (SQS), where a downstream consumer can aggregate it. The identifiers are synthetic, and the example makes no claim about measured latency or delivery reliability.

Capture usage after a completed call

LangChain’s Bedrock integration exposes token counts on AIMessage.usage_metadata. I prefer the callback boundary because the application can attach its own request context in one place. This example uses ChatBedrockConverse and a callback attached to one synchronous invocation. Check that metadata is present for the model and package versions you use.

I keep the application request ID separate from the callback run ID. The first groups model calls that belong to one application request; the second gives each usage event an identity for deduplication. requestedModelId records the configured model or inference profile identifier, which may differ from the resolved model identity.

The example uses Python with langchain-aws and boto3. Save it as token_usage.py, then set AWS_REGION, BEDROCK_MODEL_ID, and SQS_QUEUE_URL for an account that can invoke the model and send to the queue. Running it makes a model call and can incur charges.

import json
import logging
import os
from datetime import datetime, timezone
from uuid import uuid4

import boto3
from langchain_aws import ChatBedrockConverse
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.outputs import ChatGeneration

log = logging.getLogger(__name__)


class TokenUsageHandler(BaseCallbackHandler):
    def __init__(self, sqs, queue_url, model_id, request_id,
                 user_id, application_id):
        self.sqs = sqs
        self.queue_url = queue_url
        self.model_id = model_id
        self.request_id = request_id
        self.user_id = user_id
        self.application_id = application_id

    def on_llm_end(self, response, *, run_id, **kwargs):
        # This example expects one chat result for one invocation.
        if len(response.generations) != 1 or not response.generations[0]:
            log.warning("Unexpected result shape for request %s", self.request_id)
            return
        generation = response.generations[0][0]
        if not isinstance(generation, ChatGeneration):
            log.warning("Missing chat result for request %s", self.request_id)
            return
        usage = generation.message.usage_metadata
        if not usage or any(usage.get(key) is None for key in
                            ("input_tokens", "output_tokens")):
            log.warning("Missing usage for request %s", self.request_id)
            return

        event = {
            "schemaVersion": 1,
            "eventId": str(run_id),
            "requestId": self.request_id,
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "userId": self.user_id,
            "applicationId": self.application_id,
            "requestedModelId": self.model_id,
            "inputTokens": usage["input_tokens"],
            "outputTokens": usage["output_tokens"],
            "usageDetails": usage,
        }
        try:
            self.sqs.send_message(
                QueueUrl=self.queue_url,
                MessageBody=json.dumps(event),
            )
        except Exception:
            # Best-effort telemetry: this event can be lost.
            log.exception("Usage delivery failed for request %s", self.request_id)


def main():
    logging.basicConfig(level=logging.INFO)
    region = os.environ["AWS_REGION"]
    model_id = os.environ["BEDROCK_MODEL_ID"]
    # Reuse clients across requests in a long-running application.
    sqs = boto3.client("sqs", region_name=region)
    chat = ChatBedrockConverse(model_id=model_id, region_name=region)
    handler = TokenUsageHandler(
        sqs=sqs,
        queue_url=os.environ["SQS_QUEUE_URL"],
        model_id=model_id,
        request_id=str(uuid4()),
        user_id="demo-user",
        application_id="demo-assistant",
    )
    response = chat.invoke(
        [("human", "Explain a message queue in one sentence.")],
        config={"callbacks": [handler]},
    )
    print(response.content)


if __name__ == "__main__":
    main()

A successful run prints the model response. If usage metadata is present and the send succeeds, the queue receives a JSON event with input and output counts.

Those are two separate outcomes. A printed response proves that the model call returned. Telemetry delivery needs its own confirmation. Treating one as evidence of the other creates the most dangerous kind of monitoring failure: a clean application path with a silent observability gap.

Define the coverage boundary

The callback runs after a successful model call. Usage becomes available after generation. Failed attempts and cancellations can leave gaps. I call this post-call usage collection. “Real-time monitoring” would overstate its coverage. Record missing usage as an observable gap; zero means known zero.

Streaming needs its own validation. Bedrock’s ConverseStream API includes usage in a metadata event. An interrupted stream can end before your application receives that information. Verify how your integration aggregates stream metadata before reusing this handler for streaming calls.

The synchronous SQS send also adds work to the invocation path. Its network time and SDK retries can delay the return to your caller. Measure that overhead.

I would keep the synchronous send while validating the event shape and failure modes because the behavior is easy to observe. Once its latency matters, I would move delivery behind a bounded in-process queue or another durable boundary. At that point, queue capacity, shutdown behavior, and process failure become part of the design.

Make delivery failures visible

For request-path telemetry, my default is to let the user request continue and make the telemetry failure loud enough for operators to see. The example only logs an SQS failure. Its scope is integration behavior. Production operation also needs a counter, durable replay mechanism, and alert.

If the data will drive chargeback or another accounting-sensitive decision, best-effort delivery is the wrong boundary. Decide where events become durable and how failed sends will be retried.

Standard SQS queues can deliver a message more than once. Make the consumer idempotent: processing the same eventId again should produce the same total. Retrying delivery of an existing event should preserve its ID; a new model invocation should have its own event.

Track missing-usage events, send failures, consumer lag, and duplicate events alongside token totals. Use opaque identifiers and omit prompts and responses from usage events unless a separate, explicit requirement calls for their collection.

Turn usage into cost estimates

Token totals are one input to cost attribution. A pricing calculation also needs the applicable model, region, pricing date, and usage categories. Preserve cache-related details when present, and check how they relate to aggregate counts before calculating charges. Treat the result as an estimate to reconcile with billing.

Start by collecting a synthetic request end to end. Then test a missing-usage response, an SQS failure, and duplicate delivery before using the resulting totals for operational decisions.