rohitg00/ai-engineering-from-scratch · error · ProtocolError

stream ended without message_stop

Error message

stream ended without message_stop

What it means

Raised by collect_stream_text when the event iterator exhausts without a message_stop event. The collector treats message_stop as proof the stream terminated cleanly; its absence means the stream was truncated by a network drop, timeout, or an incomplete capture.

Source

Thrown at certifications/claude/lessons/08-messages-api-and-application-lifecycle/code/main.py:174

    return "".join(str(block.get("text", "")) for block in blocks if block["type"] == "text")


def collect_stream_text(events: Iterable[dict[str, Any]]) -> str:
    """Collect only text deltas while checking that a stream terminates."""
    chunks: list[str] = []
    stopped = False
    for event in events:
        event_type = event.get("type")
        if stopped:
            raise ProtocolError("event arrived after message_stop")
        if event_type == "content_block_delta":
            delta = event.get("delta", {})
            if delta.get("type") == "text_delta":
                chunks.append(str(delta.get("text", "")))
        elif event_type == "message_stop":
            stopped = True
    if not stopped:
        raise ProtocolError("stream ended without message_stop")
    return "".join(chunks)


def batch(items: list[Any], size: int) -> list[list[Any]]:
    if size < 1:
        raise ValueError("batch size must be positive")
    return [items[index : index + size] for index in range(0, len(items), size)]


def stable_cache_key(model: str, stable_prefix: str) -> str:
    payload = f"{model}\0{stable_prefix}".encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


IMAGE_MEDIA_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
DOCUMENT_MEDIA_TYPES = {"application/pdf", "text/plain"}

View on GitHub (pinned to 39ea8a1c6d)

Solutions

1. Append {"type":"message_stop"} to mock/test event lists
2. On live streams, retry the request when this error fires instead of using partial text
3. Verify captures include the final SSE event before replaying them

Example fix

// before
events = [{"type":"content_block_delta","delta":{"type":"text_delta","text":"hi"}}]
// after
events = [
  {"type":"content_block_delta","delta":{"type":"text_delta","text":"hi"}},
  {"type":"message_stop"}
]

def stream_is_complete(events):
    return bool(events) and events[-1].get("type") == "message_stop"

try:
    text = collect_stream_text(events)
except ProtocolError as exc:
    if "without message_stop" in str(exc):
        text = collect_stream_text(stream_request_again())  # discard partial

When it happens

Trigger: Passing a list of content_block_delta events with no trailing {"type":"message_stop"}, e.g. a capture that stopped mid-response (test_stream_collector_requires_stop).

Common situations: Reading a stream that hit a connection reset or client read timeout, saving a partial SSE log, or forgetting to append the terminal event in mock streams.

Related errors

• event arrived after message_stop
• tool_use requires name and object input
• response content must be a non-empty block list
• every content block needs a type
• batch size must be positive

AI-assisted analysis of rohitg00/ai-engineering-from-scratch@39ea8a1c6d (2026-08-26). Data as JSON: /api/errors/c4b2cef6a7146470. Report an issue: GitHub.