feat: naive token estimation via tiktoken

Author: lizradway

pyproject.toml

@@ -40,6 +40,7 @@ dependencies = [
     "opentelemetry-api>=1.30.0,<2.0.0",
     "opentelemetry-sdk>=1.30.0,<2.0.0",
     "opentelemetry-instrumentation-threading>=0.51b0,<1.00b0",
+    "tiktoken>=0.7.0,<1.0.0",
 ]
 
 

src/strands/models/model.py

@@ -1,16 +1,18 @@
 """Abstract base class for Agent model providers."""
 
 import abc
+import json
 import logging
 from collections.abc import AsyncGenerator, AsyncIterable
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Literal, TypeVar
 
+import tiktoken
 from pydantic import BaseModel
 
 from ..hooks.events import AfterInvocationEvent
 from ..plugins.plugin import Plugin
-from ..types.content import Messages, SystemContentBlock
+from ..types.content import ContentBlock, Messages, SystemContentBlock
 from ..types.streaming import StreamEvent
 from ..types.tools import ToolChoice, ToolSpec
 
@@ -21,6 +23,79 @@
 
 T = TypeVar("T", bound=BaseModel)
 
+_DEFAULT_ENCODING = "cl100k_base"
+
+
+def _count_content_block_tokens(block: ContentBlock, encoding: tiktoken.Encoding) -> int:
+    """Count tokens for a single content block."""
+    total = 0
+
+    if "text" in block:
+        total += len(encoding.encode(block["text"]))
+
+    if "toolUse" in block:
+        try:
+            total += len(encoding.encode(json.dumps(block["toolUse"])))
+        except (TypeError, ValueError):
+            pass
+
+    if "toolResult" in block:
+        try:
+            total += len(encoding.encode(json.dumps(block["toolResult"])))
+        except (TypeError, ValueError):
+            pass
+
+    if "reasoningContent" in block:
+        reasoning = block["reasoningContent"]
+        if "reasoningText" in reasoning:
+            reasoning_text = reasoning["reasoningText"]
+            if "text" in reasoning_text:
+                total += len(encoding.encode(reasoning_text["text"]))
+
+    if "guardContent" in block:
+        guard = block["guardContent"]
+        if "text" in guard:
+            total += len(encoding.encode(guard["text"]["text"]))
+
+    if "citationsContent" in block:
+        citations = block["citationsContent"]
+        for item in citations.get("content", []):
+            if "text" in item:
+                total += len(encoding.encode(item["text"]))
+
+    return total
+
+
+def _estimate_tokens_with_tiktoken(
+    messages: Messages,
+    tool_specs: list[ToolSpec] | None = None,
+    system_prompt: str | None = None,
+    encoding_name: str = _DEFAULT_ENCODING,
+) -> int:
+    """Estimate tokens by serializing messages/tools to text and counting with tiktoken.
+
+    This is a best-effort fallback for providers that don't expose native counting.
+    Accuracy varies by model but is sufficient for threshold-based decisions.
+    """
+    encoding = tiktoken.get_encoding(encoding_name)
+    total = 0
+
+    if system_prompt:
+        total += len(encoding.encode(system_prompt))
+
+    for message in messages:
+        for block in message["content"]:
+            total += _count_content_block_tokens(block, encoding)
+
+    if tool_specs:
+        for spec in tool_specs:
+            try:
+                total += len(encoding.encode(json.dumps(spec)))
+            except (TypeError, ValueError):
+                pass
+
+    return total
+
 
 @dataclass
 class CacheConfig:
@@ -130,6 +205,27 @@ def stream(
         """
         pass
 
+    def estimate_tokens(
+        self,
+        messages: Messages,
+        tool_specs: list[ToolSpec] | None = None,
+        system_prompt: str | None = None,
+    ) -> int:
+        """Estimate token count for the given input before sending to the model.
+
+        Used for proactive context management (e.g., triggering compression at a
+        threshold). Accuracy within 5-10% is sufficient — this is not used for billing.
+
+        Args:
+            messages: List of message objects to estimate tokens for.
+            tool_specs: List of tool specifications to include in the estimate.
+            system_prompt: System prompt to include in the estimate.
+
+        Returns:
+            Estimated total input tokens.
+        """
+        return _estimate_tokens_with_tiktoken(messages, tool_specs, system_prompt)
+
 
 class _ModelPlugin(Plugin):
     """Plugin that manages model-related lifecycle hooks."""

tests/strands/models/test_model.py

@@ -213,3 +213,151 @@ def test_model_plugin_preserves_messages_when_not_stateful(model_plugin):
     model_plugin._on_after_invocation(event)
 
     assert len(agent.messages) == 1
+
+
+def test_estimate_tokens_empty_messages(model):
+    assert model.estimate_tokens(messages=[]) == 0
+
+
+def test_estimate_tokens_system_prompt_only(model):
+    result = model.estimate_tokens(messages=[], system_prompt="You are a helpful assistant.")
+    assert result > 0
+
+
+def test_estimate_tokens_text_messages(model, messages):
+    result = model.estimate_tokens(messages=messages)
+    assert result > 0
+
+
+def test_estimate_tokens_with_tool_specs(model, messages, tool_specs):
+    without_tools = model.estimate_tokens(messages=messages)
+    with_tools = model.estimate_tokens(messages=messages, tool_specs=tool_specs)
+    assert with_tools > without_tools
+
+
+def test_estimate_tokens_with_system_prompt(model, messages, system_prompt):
+    without_prompt = model.estimate_tokens(messages=messages)
+    with_prompt = model.estimate_tokens(messages=messages, system_prompt=system_prompt)
+    assert with_prompt > without_prompt
+
+
+def test_estimate_tokens_combined(model, messages, tool_specs, system_prompt):
+    result = model.estimate_tokens(messages=messages, tool_specs=tool_specs, system_prompt=system_prompt)
+    assert result > 0
+
+
+def test_estimate_tokens_tool_use_block(model):
+    messages = [
+        {
+            "role": "assistant",
+            "content": [
+                {
+                    "toolUse": {
+                        "toolUseId": "123",
+                        "name": "my_tool",
+                        "input": {"query": "test"},
+                    }
+                }
+            ],
+        }
+    ]
+    result = model.estimate_tokens(messages=messages)
+    assert result > 0
+
+
+def test_estimate_tokens_tool_result_block(model):
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "toolResult": {
+                        "toolUseId": "123",
+                        "content": [{"text": "tool output here"}],
+                        "status": "success",
+                    }
+                }
+            ],
+        }
+    ]
+    result = model.estimate_tokens(messages=messages)
+    assert result > 0
+
+
+def test_estimate_tokens_reasoning_block(model):
+    messages = [
+        {
+            "role": "assistant",
+            "content": [
+                {
+                    "reasoningContent": {
+                        "reasoningText": {
+                            "text": "Let me think about this step by step.",
+                        }
+                    }
+                }
+            ],
+        }
+    ]
+    result = model.estimate_tokens(messages=messages)
+    assert result > 0
+
+
+def test_estimate_tokens_skips_binary_content(model):
+    messages = [
+        {
+            "role": "user",
+            "content": [{"image": {"format": "png", "source": {"bytes": b"fake image data"}}}],
+        }
+    ]
+    assert model.estimate_tokens(messages=messages) == 0
+
+
+def test_estimate_tokens_tool_result_with_bytes(model):
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "toolResult": {
+                        "toolUseId": "123",
+                        "content": [{"image": {"format": "png", "source": {"bytes": b"image data"}}}],
+                        "status": "success",
+                    }
+                }
+            ],
+        }
+    ]
+    result = model.estimate_tokens(messages=messages)
+    assert result == 0
+
+
+def test_estimate_tokens_citations_block(model):
+    messages = [
+        {
+            "role": "assistant",
+            "content": [
+                {
+                    "citationsContent": {
+                        "content": [{"text": "According to the document, the answer is 42."}],
+                        "citations": [],
+                    }
+                }
+            ],
+        }
+    ]
+    result = model.estimate_tokens(messages=messages)
+    assert result > 0
+
+
+def test_estimate_tokens_all_inputs(model):
+    messages = [
+        {"role": "user", "content": [{"text": "hello world"}]},
+        {"role": "assistant", "content": [{"text": "hi there"}]},
+    ]
+    result = model.estimate_tokens(
+        messages=messages,
+        tool_specs=[{"name": "test", "description": "a test tool", "inputSchema": {"json": {}}}],
+        system_prompt="Be helpful.",
+    )
+    assert result > 0