Preventing Token Overflow in Production Claude Apps

Token overflow is the failure mode nobody plans for and everyone eventually hits. Not the 400 error — the slow drift toward it. Your Claude app ships with generous margins, works fine for a month, then one Tuesday you notice conversations getting sluggish, bills getting steeper, and users mentioning the model "forgot" what they said. By the time you see the actual overflow error, you've been leaking cost and quality for weeks. This post is the shape of what to build so you don't.

The pattern that follows is the practical distillation. It covers what to count, how to count it accurately, how to set a budget, how to compress conversation history when the budget gets tight, and how to compress the tool results that quietly eat most of your budget in agent-style apps. Everything is Claude-specific — the count_tokens API, the 200K Sonnet window, the tool_use format, the cache interaction. The complete implementation is a class you can drop into any service.

What "token overflow" actually means

Everyone thinks of overflow as the hard error — the 400 that says your prompt exceeded the context window. That's the visible tip. The invisible costs arrive much earlier, and they compound.

The naive response to overflow is to raise the model's window or wait for larger context models. Neither is the answer. Even with a million-token window, the quality drop hits well before you fill it. What you actually want is deliberate budget management — treating tokens as a scarce resource you allocate on purpose, not a limit you hit by accident.

The four consumers of your context budget

Every Claude request's input is the sum of four buckets. Optimizing without knowing which one is dominating is guessing.

# Every Claude request's total input tokens =
input_tokens = (
    system_prompt_tokens         # system message + persona
    + tool_definition_tokens     # each tool's schema costs tokens
    + conversation_history_tokens # all prior user/assistant turns
    + current_user_message_tokens # what the user just said
    + attachment_tokens           # images, PDFs, files
)

# Plus your reservation for the response:
total_budget = input_tokens + max_tokens

# This total must fit within the model's context window.
# Claude Sonnet 4.6: 200,000 tokens
# Claude Opus 5:     200,000 tokens
# Claude Haiku 4.5:  200,000 tokens

Instrumenting each bucket separately tells you where to optimize. If system + tools is 8K and conversation history is 60K, compressing history is 7× more valuable than trimming your system prompt. Most teams optimize the wrong bucket first because they never measured which one dominated.

Counting tokens accurately — the count_tokens API

You can't manage what you don't measure, and rough estimates are wrong more often than they're right. Anthropic provides a count_tokens endpoint that returns exact input token counts for any message array. It's free (not billed against your account) and it handles multimodal content, tool definitions, and system prompts correctly.

from anthropic import Anthropic

client = Anthropic()

def count_claude_tokens(
    system: str | None = None,
    messages: list | None = None,
    tools: list | None = None,
) -> int:
    # Exact input token count via Anthropic's free count_tokens API.
    # Handles multimodal (images, PDFs), tool definitions, and system prompts.
    kwargs = {
        "model": "claude-sonnet-4-6",
        "messages": messages or [],
    }
    if system:
        kwargs["system"] = system
    if tools:
        kwargs["tools"] = tools
    result = client.messages.count_tokens(**kwargs)
    return result.input_tokens


# Fast path for hot loops: local estimate
import tiktoken

_encoder = tiktoken.get_encoding("cl100k_base")

def fast_estimate(text: str) -> int:
    # Rough proxy for Claude's tokenizer. Usually within 5-10% for English.
    # Use for pre-flight budget checks; verify with count_tokens if close.
    return int(len(_encoder.encode(text)) * 1.1)

Two counters: exact and fast. Use exact for anything that hits budget boundaries. Use fast for the hot path where a millisecond matters more than a few percent. When fast estimates get close to your budget threshold, verify with exact.

The budget model

The single most useful concept in this post is the budget model. Instead of "how much fits?", ask "how do I allocate a fixed budget?" — and enforce it before every call.

from dataclasses import dataclass

@dataclass
class ContextBudget:
    total_window: int              # model's context window
    reserved_output: int           # max_tokens for the response
    reserved_system: int           # system prompt + tools
    reserved_current_turn: int     # user's current message
    safety_margin_pct: float = 0.05
    
    @property
    def safety_margin(self) -> int:
        return int(self.total_window * self.safety_margin_pct)
    
    @property
    def available_for_history(self) -> int:
        # How much room is left for conversation history + retrieval
        return (self.total_window
                - self.reserved_output
                - self.reserved_system
                - self.reserved_current_turn
                - self.safety_margin)


# Typical config for a chat app on Claude Sonnet 4.6
budget = ContextBudget(
    total_window=200_000,        # Sonnet's window
    reserved_output=4_000,       # max response length
    reserved_system=2_000,       # system prompt + tool defs
    reserved_current_turn=1_000, # user's current message
)
# Available for history: ~183,000 tokens
# Safety margin: 10,000 tokens (5%)
# Effective history budget: ~173,000 tokens

The safety margin catches tokenizer variance and estimation error. Without it, an off-by-a-few-tokens counting error causes hard failures at the boundary. Five percent is a comfortable buffer without being wasteful. Right-size reserved_output to your actual p99 response length — if your responses are always under 1K, reserving 4K wastes 3K of history budget every request.

Compressing conversation history

Long conversations are the biggest consumer of context in most chat apps. A hundred-turn conversation with 500-token average messages already costs 50K tokens in history alone. Compression is how you keep it manageable.

Strategy 1: Sliding window (simplest)

def sliding_window(messages: list, keep_turns: int = 20) -> list:
    # Keep the first message (often has important context) + last N turns.
    # Drops everything in the middle.
    if len(messages) <= keep_turns:
        return messages
    return messages[:1] + messages[-keep_turns:]

Simple, deterministic, no LLM call. Works well for chat apps with strong recency bias — what was just said matters more than what was said an hour ago. Fails when users refer back to something ancient and the model "forgets."

Strategy 2: Rolling summary

from anthropic import Anthropic
client = Anthropic()

def rolling_summary(
    messages: list,
    threshold_tokens: int = 20_000,
    keep_recent: int = 10,
) -> list:
    # Periodically summarize older turns into a single summary message.
    # Preserves salient info at much lower token cost.
    total = count_claude_tokens(messages=messages)
    if total < threshold_tokens:
        return messages
    
    to_summarize = messages[:-keep_recent]
    verbatim = messages[-keep_recent:]
    if not to_summarize:
        return messages
    
    # Use Haiku for compression — cheap and fast
    prompt_text = "\n\n".join(
        f"{m['role']}: {m.get('content', '')}" for m in to_summarize
    )
    summary = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1500,
        system=("Summarize this conversation in 3-5 paragraphs. Preserve: "
                "user goals, decisions made, personal details shared, "
                "unresolved questions. Omit: pleasantries, filler."),
        messages=[{"role": "user", "content": prompt_text}],
    )
    summary_text = summary.content[0].text
    
    return [
        {"role": "user", "content": "Earlier conversation summary:"},
        {"role": "assistant", "content": summary_text},
    ] + verbatim

Preserves salient information at much lower token cost. Uses Haiku (~$0.80 per million input) for the compression call, which is negligible compared to the tokens saved on subsequent Sonnet requests. The tradeoff is a small LLM call every time compression fires — but it fires infrequently (only when the threshold is crossed), so amortized cost is low.

The interaction with prompt caching: re-summarizing on every request breaks Anthropic's prompt cache. If you want to preserve cache benefits, only re-summarize when the conversation has doubled since the last summary, not on every turn. The compressed message array plus the recent turns can then be a stable prefix that hits the cache repeatedly.

Tool result compression — the hidden context eater

In agent-style Claude apps — the ones using tool_use extensively — tool results dominate the context budget. A single database query result, file read, or web search response is often 5K-20K tokens. After a dozen tool calls, the conversation is mostly tool outputs, most of which the model no longer needs to reason about.

def compress_tool_result(content: str, max_tokens: int = 2000) -> str:
    # Truncate long tool results with an explicit marker.
    # Applied at tool result creation time.
    tokens = fast_estimate(content)
    if tokens <= max_tokens:
        return content
    
    ratio = max_tokens / tokens
    keep_chars = int(len(content) * ratio)
    truncated = content[:keep_chars]
    
    # For JSON, try to end at a valid delimiter
    for delim in ['}\n', ',\n', ',', '}']:
        cut = truncated.rfind(delim)
        if cut > keep_chars * 0.9:
            truncated = truncated[:cut + len(delim)]
            break
    
    return (
        truncated
        + f"\n\n[Truncated: showing first ~{max_tokens} of {tokens} tokens]"
    )


def age_out_old_tool_results(
    messages: list,
    keep_recent: int = 5,
) -> list:
    # Replace old tool_result content with a compact marker.
    # Most impactful optimization for long agent conversations.
    tool_result_locations = []
    for i, m in enumerate(messages):
        if isinstance(m.get("content"), list):
            for j, block in enumerate(m["content"]):
                if block.get("type") == "tool_result":
                    tool_result_locations.append((i, j))
    
    # Keep the most recent N; age out the rest
    to_age = tool_result_locations[:-keep_recent] \
             if len(tool_result_locations) > keep_recent else []
    
    for i, j in to_age:
        block = messages[i]["content"][j]
        content = block.get("content", "")
        if isinstance(content, str) and len(content) > 200:
            tokens = fast_estimate(content)
            block["content"] = f"[Aged out tool result, was ~{tokens} tokens]"
    return messages

Two techniques. The first (compress_tool_result) truncates at insertion — every tool result over 2K tokens gets a summary marker before it enters the context. The second (age_out_old_tool_results) rewrites older tool results in place, keeping recent ones verbatim. Together they can 3-5× the effective conversation length before other context management kicks in.

The tool_use / tool_result pairing rule: Claude requires every tool_use in the conversation to have a matching tool_result before the next assistant turn. Don't drop tool_result blocks entirely — that breaks the pairing and every subsequent request fails with a 400. Only ever replace the content field with a marker; never remove the block itself.

Putting it together: the ContextManager

Composed into one class. Drop it into your service and every Claude call goes through it.

from typing import Callable

class ContextManager:
    def __init__(
        self,
        budget: ContextBudget,
        summarizer: Callable | None = None,
        tool_result_max_tokens: int = 2000,
        keep_recent_tool_results: int = 5,
        summary_threshold_pct: float = 0.7,
        keep_recent_turns: int = 10,
    ):
        self.budget = budget
        self.summarizer = summarizer
        self.tool_result_max = tool_result_max_tokens
        self.keep_tool_results = keep_recent_tool_results
        self.summary_threshold_pct = summary_threshold_pct
        self.keep_recent = keep_recent_turns
    
    def prepare(
        self,
        system: str,
        tools: list,
        history: list,
        current_message: dict,
    ) -> dict:
        # Step 1: age out old tool results
        history = age_out_old_tool_results(
            history, keep_recent=self.keep_tool_results
        )
        
        # Step 2: measure current history size
        history_tokens = count_claude_tokens(messages=history)
        
        # Step 3: compress if history exceeds threshold
        threshold = int(
            self.budget.available_for_history * self.summary_threshold_pct
        )
        if history_tokens > threshold and self.summarizer:
            history = self.summarizer(history, keep_recent=self.keep_recent)
            history_tokens = count_claude_tokens(messages=history)
        
        # Step 4: if still over, truncate as last resort
        if history_tokens > self.budget.available_for_history:
            history = self._truncate_oldest(
                history, self.budget.available_for_history
            )
        
        # Step 5: log utilization
        total_input = (
            count_claude_tokens(system=system, tools=tools) + history_tokens
        )
        utilization_pct = (total_input / self.budget.total_window) * 100
        
        return {
            "system": system,
            "tools": tools,
            "messages": history + [current_message],
            "metadata": {
                "input_tokens": total_input,
                "history_tokens": history_tokens,
                "utilization_pct": utilization_pct,
            }
        }
    
    def _truncate_oldest(self, messages: list, target: int) -> list:
        # Drop oldest pairs until we fit. Keep first message + recent turns.
        while count_claude_tokens(messages=messages) > target:
            if len(messages) <= 4:
                break
            messages = messages[:1] + messages[3:]  # drop pair after intro
        return messages


# Usage
def haiku_summarizer(messages: list, keep_recent: int = 10) -> list:
    return rolling_summary(messages, keep_recent=keep_recent)

budget = ContextBudget(
    total_window=200_000,
    reserved_output=4_000,
    reserved_system=2_000,
    reserved_current_turn=1_000,
)

ctx_mgr = ContextManager(
    budget=budget,
    summarizer=haiku_summarizer,
    tool_result_max_tokens=2000,
    keep_recent_tool_results=5,
    summary_threshold_pct=0.7,
)

# In your request handler
prepared = ctx_mgr.prepare(
    system=SYSTEM_PROMPT,
    tools=MY_TOOLS,
    history=conversation_history,
    current_message={"role": "user", "content": user_input},
)

if prepared["metadata"]["utilization_pct"] > 90:
    logger.warning(f"High context util: {prepared['metadata']['utilization_pct']:.0f}%")

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=budget.reserved_output,
    system=prepared["system"],
    tools=prepared["tools"],
    messages=prepared["messages"],
)

What to monitor

Instrument utilization on every request. Context overflow is preceded by weeks of measurable drift — the utilization graph rises 2-3% per week as conversations get longer, tool results grow, or the corpus expands. Catch it on the dashboard, not in the error logs.

from prometheus_client import Histogram, Counter

context_utilization = Histogram(
    "claude_context_utilization_pct",
    "Input tokens as % of Claude's context window",
    buckets=[10, 30, 50, 70, 85, 95, 99],
)

overflow_events = Counter(
    "claude_overflow_events_total",
    "Times context exceeded budget (compression fired or hard error)",
    ["reason"],  # reason: compression, truncation, hard_error
)

# Wire into ContextManager.prepare
def emit_metrics(prepared: dict):
    context_utilization.observe(prepared["metadata"]["utilization_pct"])
    if prepared["metadata"].get("compression_applied"):
        overflow_events.labels(
            reason=prepared["metadata"]["compression_applied"]
        ).inc()

Set alerts at 70% (capacity planning window) and 90% (imminent overflow). If your p95 utilization sits above 70% for a week, plan a compression tightening or a per-workload budget adjustment before the p95 becomes p99 and users see it.

What this doesn't do

The version above is the pragmatic minimum. For higher-scale services, extensions worth adding:

• Prompt caching optimization — Anthropic's cache_control markers on stable prefixes (system prompt, tool defs, long documents) can reduce cost 60-90%. The compression pattern here doesn't break the cache if you re-summarize infrequently.
• RAG for very long documents — if you're stuffing multi-hundred-K documents into every request, retrieval beats stuffing on both cost and quality. Chapter 7 of the pillar guide covers when to make the switch.
• Persistent summary across sessions — store the summary in your database keyed by conversation ID. When the user returns, load it and prepend. Long-term memory without exploding context.
• Distributed state — the summary cache is per-instance. For multi-instance apps, back it with Redis.
• Multi-modal token counting — images add tokens too (~1,600 per image on Claude at high res). Extend the count for image attachments.

Where to go deeper

Each layer here has depth worth understanding when you're scaling. The pillar guide covers the complete pattern:

• Preventing Context Window Overflow — Production Patterns — the full treatment: token counting mechanics, the "lost in the middle" phenomenon, RAG vs stuffing decisions, hierarchical compression, prompt caching interactions, complete ContextManager with all strategies.
• How to Build a Bulletproof Claude API Wrapper — pairs with this post: context management above, reliability primitives below. Two orthogonal concerns, one composed service.
• Handling Rate Limits Across AI APIs — smaller context means fewer tokens which means less TPM pressure. Context management is indirect rate limit management.
• Fallback Strategies: Claude → GPT → Gemini — context prep must happen before the fallback layer. This ContextManager output feeds directly into the FallbackChain input.

Frequently asked questions