Claude API Streaming Breaks the "Everything Arrives" Assumption — Field Notes on Recovering from Partial Failure

When cut off, hand back "this far" — not "resume" — and let it write on

When the connection dropped, the first thing I considered was whether I could resume the same request. But Claude API has no per-event replay mechanism, so reconnecting to the same call gives no guarantee about where it picks up. You might get duplicates, or it might start from the top. That makes deterministic reassembly impossible.

What is reliable is to push the partial response you have received so far into the conversation history as an assistant message and have it generate the continuation. The request count goes up, but because each round is determined, you can treat it idempotently. Below is a thin wrapper that catches only connection-class errors and reconnects automatically. The deliberate choice not to handle rate limits here matters, for reasons I cover later.

import asyncio, logging
import anthropic
 
log = logging.getLogger(__name__)
 
class ResilientStream:
    def __init__(self, client: anthropic.AsyncAnthropic, max_retries: int = 3):
        self.client = client
        self.max_retries = max_retries
        self.state = StreamState()
 
    async def run(self, **kwargs):
        attempt = 0
        while True:
            try:
                async with self.client.messages.stream(**kwargs) as stream:
                    async for event in stream:
                        self._track(event)
                        yield event
                    return  # reached message_stop
            except (anthropic.APIConnectionError, anthropic.APITimeoutError) as e:
                attempt += 1
                if attempt > self.max_retries:
                    log.error("stream gave up after %d attempts: %s", attempt, e)
                    raise
                await asyncio.sleep(min(2 ** attempt, 20))
                kwargs = self._resume_kwargs(kwargs)  # push partial response into history
 
    def _track(self, event) -> None:
        t = event.type
        if t == "message_start":
            self.state.message_id = event.message.id
            self.state.model = event.message.model
        elif t == "content_block_start":
            self.state.current_index = event.index
        elif t == "content_block_delta":
            self.state.apply_delta(event.index, event.delta.model_dump())
        elif t == "message_delta" and event.delta.stop_reason:
            self.state.stop_reason = event.delta.stop_reason
 
    def _resume_kwargs(self, original: dict) -> dict:
        partial = self.state.assistant_text()
        if not partial:
            return original
        resumed = dict(original)
        resumed["messages"] = list(original.get("messages", [])) + [
            {"role": "assistant", "content": partial}
        ]
        return resumed

From the caller's side, no matter how many times it was cut off, it looks like a complete response flowed through. Because all the recovery logic hides inside this boundary, the rendering and business logic need to know nothing about it. How you draw that boundary pays off later.

Don't double the text when the same delta arrives twice

Duplication fails loudly. The moment text is concatenated twice, the user notices on sight. Depending on HTTP/2 retries and how reconnection is implemented, the same content_block_delta flowing again is perfectly ordinary, so dedup cannot be skipped.

The crucial part here is being explicit about what counts as identical. Timestamps can change on resend, so they cannot be the key. Where I settled was combining three things: the message ID, the block index, and the cumulative receive position within that block. Including the position also prevents the accident of mistaking a different delta of the same block for one another.

from hashlib import sha256
 
class DedupWindow:
    """Remember the last `window` events and drop anything seen a second time."""
    def __init__(self, window: int = 512):
        self.seen: set[str] = set()
        self.order: list[str] = []
        self.window = window
 
    def is_duplicate(self, message_id: str, index: int, position: int) -> bool:
        key = sha256(f"{message_id}:{index}:{position}".encode()).hexdigest()
        if key in self.seen:
            return True
        self.seen.add(key)
        self.order.append(key)
        if len(self.order) > self.window:
            self.seen.discard(self.order.pop(0))
        return False

To keep memory from growing without bound, I cap how many keys to remember. A single response holds at most a few hundred deltas, so 512 entries amply cover one round. Without that cap, the set creeps upward on a server left running for a long time.

Don't let half-finished tool arguments touch production data

The one that froze my blood was not garbled text but a tool call. Tool arguments arrive as JSON string fragments via input_json_delta. If cut off mid-way, that fragment stays as broken JSON. Trying to parse it and getting an exception is the lucky case; the worst is when it happens to be valid JSON up to the cut point, and the tool runs with half its arguments.

It actually happened once that a write tool ran with a parameter missing from a mid-stream disconnect, and nearly applied an unintended update to a production record. I caught it just in time, but ever since, I have replaced the very condition for executing a tool with "the arguments parse completely."

import json
from typing import Optional
 
def parse_tool_input(block: dict) -> Optional[dict]:
    """Safely read a tool_use block's partial_json. Return None if incomplete."""
    raw = block.get("partial_json", "").strip()
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        log.warning("incomplete tool input, skipping: %s (head=%r)", e, raw[:120])
        return None
 
# Always pass through this at the execution boundary
args = parse_tool_input(block)
if args is None:
    # Do not start the tool while arguments are incomplete
    return
run_tool(block["name"], args)

It is only a few lines, but placing this one check at the application boundary nearly closes the path by which a mid-stream interruption turns into a side effect. A safety valve works even when it is unglamorous.

Rest connection errors and rate limits on separate rhythms

There was a reason the wrapper deliberately did not handle rate limits. If you run connection-error retries and rate-limit retries through the same machinery, they interfere and retries pile up like a storm. Re-establishing the connection at short intervals while 429s are coming back simply invites more 429s.

So vary how you rest by error type. For rate limits, respect the wait time the server returns and top up the shortfall with jitter. For connection-class errors, use a shorter exponential backoff. The jitter is there to scatter the phenomenon of many clients trying to recover at once and forming a fresh peak; shifting things by just a few seconds visibly smooths the simultaneous-recovery spike.

import random
 
async def backoff_for(attempt: int, error: Exception) -> None:
    if isinstance(error, anthropic.RateLimitError):
        base = getattr(error, "retry_after", None) or 30
        delay = base + random.uniform(0, 5)         # server hint + scatter
    elif isinstance(error, (anthropic.APIConnectionError, anthropic.APITimeoutError)):
        delay = min(2 ** attempt, 20) + random.uniform(0, 1)
    else:
        delay = min(2 ** attempt, 60)
    log.info("backoff %.1fs (attempt=%d, %s)", delay, attempt, type(error).__name__)
    await asyncio.sleep(delay)

Once I separated the types, retries stopped cascading into a self-inflicted choke even during high-load windows. A retry only becomes a recovery strategy once its interval is part of the design.

Don't miss the error event that arrives as a 200

There is one more step you can trip over when handling HTTP yourself. Claude API's error event arrives embedded in the SSE body while the HTTP status stays 200. The official SDK converts it to an exception, but if you write the low layer yourself, the error slips through disguised as a normal response unless you check event.type == "error" on your own.

if event.type == "error":
    err = event.error
    if err.type == "overloaded_error":
        raise anthropic.APIStatusError(err.message)  # capacity issues are worth a retry
    raise RuntimeError(f"fatal stream error: {err.type}: {err.message}")

Here only overloaded_error has a realistic chance of recovery, so route it to a retry and treat everything else as failure. Without that distinction, a defect on the prompt side quietly builds a loop that retries forever.

Catch the signs of breakage before your users do

Hardening the implementation means little if you cannot notice degradation. The worst state is the one where the user is the first to notice. In my own service, I watch these four.

Completion rate is the overall health score, reconnects reflect path instability, first-token latency reflects perceived speed, and the duplicate rate reflects an anomaly in the delivery path. The moment any single one crosses its line, you can act before a user asks "isn't this slow lately?" Monitoring is the last layer of recovery.

Roll it in three steps, in order

Trying to fold in everything at once makes it hard to start. I introduced it in this order: first replace one existing stream with the stateful wrapper, then add dedup, and finally slot the tool-argument safety valve into the execution boundary. Monitoring can come after that.

There is a reason for the order: once the stateful wrapper is in place, both dedup and the safety valve can be written by just "looking at the current state." Building from the foundation makes the latter two come out in surprisingly short code. Rather than aiming never to break, remember where you broke and continue quietly. Just re-framing it that way genuinely shifts the felt stability by a notch under the same traffic. The stretch of time during which users stop noticing the glitches is, I think, the return this design pays back.