Stream Long Responses

What Counts as a Long Response

Most answers are a few hundred tokens. Some tasks produce far more: full tutorials, reports, large code files, documentation, or translations of long documents. These long responses, from several thousand up to tens of thousands of tokens, raise problems that short answers do not:

This lesson covers techniques for each problem.

Technique 1: Always Stream Long Requests

Streaming keeps the connection active, with events flowing continuously, which avoids timeouts. It is required for very large max_tokens values (Lesson 13). Even when you do not display anything, stream and collect the final message:

from anthropic import Anthropic

client = Anthropic()

with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=64000,
    messages=[{"role": "user", "content": "Write a complete beginner's guide to Python testing."}],
) as stream:
    message = stream.get_final_message()

print(message.stop_reason, message.usage.output_tokens)

According to the streaming documentation, this pattern is especially useful for large max_tokens values, where the SDK requires streaming to avoid HTTP timeouts.

Output Limits

Remember that thinking counts toward max_tokens (Lesson 13), so leave headroom above the expected length of the visible text.

Technique 2: Show Progress

For long outputs, a running indicator reassures users that work is progressing:

written = 0
with client.messages.stream(model="claude-opus-5-5", max_tokens=32000, messages=messages) as stream:
    for text in stream.text_stream:
        written += len(text)
        print(f"\rGenerating... {written:,} characters", end="", flush=True)
    message = stream.get_final_message()
print()

\r moves the cursor back to the start of the line, so the counter updates in place. {written:,} adds thousands separators, as in 12,480.

Technique 3: Write Output Progressively

Writing each chunk to a file as it arrives means the work is preserved if the program stops unexpectedly:

with open("guide.md", "w", encoding="utf-8") as file, client.messages.stream(
    model="claude-opus-5-5", max_tokens=32000, messages=messages
) as stream:
    for text in stream.text_stream:
        file.write(text)
        file.flush()

If a long stream is interrupted, the file contains everything received so far, which can then be continued with the technique from Lesson 57.

Technique 4: Generate in Sections

For very long documents, a single giant response is rarely the best approach. Generating section by section has major advantages:

The pattern is a form of prompt chaining (Lesson 28):

1. Generate an outline (short request, structured output)
2. For each section: stream the section, with the outline as context
3. Combine the sections into the final document

Including the full outline in each section's prompt keeps sections consistent and prevents them from repeating each other's content.

Technique 5: Handle Truncation and Interruptions

Even with careful limits, a long section can stop at max_tokens, or a stream can be interrupted:

Approaches Compared

Practical Example: A Section-by-Section Document Generator

This script generates a long tutorial in three stages: an outline with structured output, each section streamed to a Markdown file with a progress display, and a summary of the work done. Each finished section is written to disk immediately, so the file is a checkpoint at every step.

"""long_document.py: generate a long document section by section with streaming."""

import time
from pathlib import Path

from anthropic import Anthropic
from pydantic import BaseModel

client = Anthropic()
MODEL = "claude-opus-5-5"
SYSTEM = "You are a technical writer creating clear, detailed tutorials for beginner Python developers."


class Outline(BaseModel):
    title: str
    sections: list[str]


def make_outline(topic: str, count: int) -> Outline:
    result = client.messages.parse(
        model=MODEL,
        max_tokens=2048,
        system=SYSTEM,
        output_format=Outline,
        messages=[{"role": "user", "content": f"Create an outline for a tutorial on {topic} "
                                              f"with exactly {count} section titles."}],
    )
    return result.parsed_output


def section_prompt(outline: Outline, index: int) -> str:
    outline_text = "\n".join(f"{n}. {title}" for n, title in enumerate(outline.sections, start=1))
    return (
        f"<outline>\n{outline_text}\n</outline>\n\n"
        f'Write section {index + 1}, "{outline.sections[index]}", of the tutorial "{outline.title}". '
        "Write about 600 words with one code example and its output. "
        "Start with a level-2 Markdown heading. Write only this section."
    )


def stream_section(outline: Outline, index: int, file) -> int:
    label = f"section {index + 1}/{len(outline.sections)}"
    written = 0
    with client.messages.stream(
        model=MODEL,
        max_tokens=8192,
        system=SYSTEM,
        messages=[{"role": "user", "content": section_prompt(outline, index)}],
    ) as stream:
        for text in stream.text_stream:
            file.write(text)
            written += len(text)
            print(f"\r  {label}: {written:,} characters", end="", flush=True)
        final = stream.get_final_message()

    file.write("\n\n")
    file.flush()
    note = "" if final.stop_reason == "end_turn" else f" [stopped: {final.stop_reason}]"
    print(f"\r  {label}: {written:,} characters, {final.usage.output_tokens} output tokens{note}")
    return final.usage.output_tokens


def generate(topic: str, sections: int, path: Path) -> None:
    start = time.perf_counter()
    outline = make_outline(topic, sections)
    print(f"Outline: {outline.title}")

    total_output = 0
    with path.open("w", encoding="utf-8") as file:
        file.write(f"# {outline.title}\n\n")
        for index in range(len(outline.sections)):
            total_output += stream_section(outline, index, file)

    minutes = (time.perf_counter() - start) / 60
    print(f"Done: {path} | {total_output:,} output tokens | {minutes:.1f} minutes")


if __name__ == "__main__":
    generate("working with files in Python", sections=4, path=Path("files_tutorial.md"))

Example output:

Outline: Working with Files in Python
  section 1/4: 4,112 characters, 1,384 output tokens
  section 2/4: 4,560 characters, 1,502 output tokens
  section 3/4: 4,021 characters, 1,349 output tokens
  section 4/4: 4,388 characters, 1,455 output tokens
Done: files_tutorial.md | 5,690 output tokens | 2.6 minutes

How It Works

1. make_outline() uses structured output (Lesson 36) to get a title and an exact list of section titles as a typed Outline object. A short, structured first step makes the rest of the process predictable.
2. section_prompt() includes the full outline in every section request, so each section knows what the others cover. The instruction "Write only this section" prevents overlap. The outline text is built in a separate variable, which keeps the f-string simple and readable.
3. stream_section() streams one section directly into the file, updating a single progress line with \r. After the stream, get_final_message() provides the stop reason and output tokens.
4. Non-normal stop reasons are reported next to the section, such as [stopped: max_tokens], so incomplete sections are visible. Exercise 2 adds automatic continuation.
5. file.flush() after each section writes completed work to disk. If the program stops during section 3, sections 1 and 2 are already saved.
6. Moderate per-section limits (8,192 tokens) are enough for about 600 words plus thinking, so no single request needs a huge max_tokens value.

Common Mistakes

Generating Long Outputs Without Streaming

What happens: timeouts, or the SDK refuses the request because it would take too long.

Fix: stream every long request, even without a display.

One Enormous Request for an Entire Document

What happens: a late failure loses everything, and long outputs drift in structure and consistency.

Fix: generate an outline, then sections.

Sections Without Shared Context

What happens: sections repeat each other or contradict each other.

Fix: include the outline, and optionally summaries of earlier sections, in every section prompt.

Ignoring Truncation in Long Sections

What happens: documents contain sections that stop mid-sentence.

Fix: check stop_reason per section, and continue or regenerate.

Keeping Output Only in Memory

What happens: a crash late in a long run loses all generated text.

Fix: write and flush output progressively.

Best Practices

• Stream every long request, using get_final_message() when no display is needed.
• Show progress for long operations.
• Write output to disk progressively, and treat each finished section as a checkpoint.
• Generate long documents section by section, with a shared outline.
• Check stop reasons per section, and continue or regenerate as needed.
• Use the Batches API for many long outputs that are not urgent.
• Leave headroom in max_tokens for thinking.

Exercise

1. Run long_document.py with a topic of your choice and 3 sections. Open the Markdown file and check that the sections do not repeat each other.
2. Add automatic continuation to stream_section(): if stop_reason is max_tokens, request a continuation (Lesson 57) and stream it into the same file.
3. Add resume support: before generating a section, check whether the file already contains its heading, and skip it if so.
4. Include a one-sentence summary of each completed section in the prompt for the next section, and compare the consistency of the results.

def stream_text(messages: list[dict], file, max_tokens: int = 8192):
    with client.messages.stream(
        model=MODEL, max_tokens=max_tokens, system=SYSTEM, messages=messages
    ) as stream:
        for text in stream.text_stream:
            file.write(text)
        return stream.get_final_message()


def stream_section_with_continuation(outline: Outline, index: int, file, max_rounds: int = 2) -> None:
    prompt = section_prompt(outline, index)
    final = stream_text([{"role": "user", "content": prompt}], file)
    written = "".join(b.text for b in final.content if b.type == "text")

    rounds = 0
    while final.stop_reason == "max_tokens" and rounds < max_rounds:
        rounds += 1
        continuation = (
            f"{prompt}\n\nYour previous response was cut off and ended with:\n"
            f"<previous_response>\n{written}\n</previous_response>\n"
            "Continue from exactly where you left off. Do not repeat earlier text."
        )
        final = stream_text([{"role": "user", "content": continuation}], file)
        written += "".join(b.text for b in final.content if b.type == "text")

    file.write("\n\n")
    file.flush()

The continuation is a user message containing the original prompt and the partial section, which follows the Claude 4.6+ pattern. max_rounds limits how many continuations are attempted, so a section can never loop forever.