Close Streams Safely

Why Closing Streams Matters

A stream holds an open HTTP connection for as long as it runs. When your program is finished with a stream, whether because it completed, failed, or is no longer needed, that connection must be closed.

If streams are not closed properly:

Closing a stream tells the API that you no longer need the response. Tokens generated before the stream was closed are still billed. Closing early limits cost; it does not make the request free.

The with Statement Does It for You

The recommended way to use streams is inside a with block. When the block ends, for any reason, the stream is closed:

with client.messages.stream(model="claude-opus-5-5", max_tokens=1024, messages=messages) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
# The stream is closed here

The block ends, and the stream is closed, in all of these cases:

Why this works: with calls the stream's cleanup method when the block exits, whether normally or because of an exception. That is the same guarantee that makes with open(...) safe for files.

Stopping a Stream Early

To stop a response deliberately, leave the with block. Use break, then let the block end:

collected = []
with client.messages.stream(
    model="claude-opus-5-5",
    max_tokens=4096,
    messages=[{"role": "user", "content": "List 100 Python tips."}],
) as stream:
    for text in stream.text_stream:
        collected.append(text)
        if len("".join(collected)) > 1000:
            break          # Leave the loop; the with block then closes the stream

print("".join(collected))

Do not call get_final_message() after deciding to stop. It would continue reading the rest of the response until it completes (Lesson 55), which defeats the purpose of stopping early.

Explicit close()

When a with block is not practical, for example when a stream object is passed between functions, close it explicitly, with try/finally to guarantee it happens:

stream = client.messages.create(
    model="claude-opus-5-5", max_tokens=1024, messages=messages, stream=True
)
try:
    for event in stream:
        ...
finally:
    stream.close()

finally runs whether the loop completes, breaks, or raises an exception, so the connection is always closed.

Closing Async Streams

The same principles apply to async streams (Lesson 56): async with closes the stream when the block exits.

async with client.messages.stream(model="claude-opus-5-5", max_tokens=1024, messages=messages) as stream:
    async for text in stream.text_stream:
        ...
# Closed here

Cancellation and Timeouts

In async code, a task can be cancelled, for example when a user disconnects or a time limit is reached. Cancellation raises asyncio.CancelledError inside the task, and async with closes the stream as the exception passes through.

A time limit for a whole streamed answer can be applied with asyncio.timeout() (Python 3.11+; on Python 3.10, use asyncio.wait_for()):

import asyncio


async def answer_with_time_limit(question: str, seconds: float) -> str:
    chunks = []
    try:
        async with asyncio.timeout(seconds):
            async with client.messages.stream(
                model="claude-opus-5-5",
                max_tokens=4096,
                messages=[{"role": "user", "content": question}],
            ) as stream:
                async for text in stream.text_stream:
                    chunks.append(text)
    except TimeoutError:
        chunks.append("\n[stopped: time limit reached]")
    return "".join(chunks)

When the time limit expires, the running code is cancelled, the inner async with closes the stream, and the outer async with converts the cancellation into TimeoutError, which is handled.

Closing Streams When Web Clients Disconnect

In a web server, a user may close the browser tab while an answer is still streaming. The server should notice and stop the stream, instead of generating text for nobody.

With FastAPI (Starlette), await request.is_disconnected() reports whether the client has gone away:

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from anthropic import AsyncAnthropic

app = FastAPI()
client = AsyncAnthropic()


@app.get("/ask")
async def ask(request: Request, q: str):
    async def events():
        async with client.messages.stream(
            model="claude-opus-5-5",
            max_tokens=2048,
            messages=[{"role": "user", "content": q}],
        ) as stream:
            async for text in stream.text_stream:
                if await request.is_disconnected():
                    break                       # Leaving async with closes the stream
                yield f"data: {text!r}\n\n"
    return StreamingResponse(events(), media_type="text/event-stream")

Checking for disconnection on each chunk means an abandoned answer is stopped within a moment of the user leaving. Lesson 56 shows a complete version with JSON-encoded events.

Generators That Wrap Streams

Applications often wrap a stream in a generator function that yields text to other code. If the consumer stops early, Python closes the generator, which raises GeneratorExit at the paused yield. A with block inside the generator then closes the stream:

def stream_text(question: str):
    with client.messages.stream(
        model="claude-opus-5-5",
        max_tokens=2048,
        messages=[{"role": "user", "content": question}],
    ) as stream:
        yield from stream.text_stream


for number, chunk in enumerate(stream_text("List 50 Python tips.")):
    print(chunk, end="")
    if number == 20:
        break      # The generator is closed, which closes the stream inside it

Why it works: when the for loop breaks, the generator object is no longer used. Python closes it, which raises GeneratorExit inside it, and the with block's cleanup runs. To make the cleanup happen at a predictable moment, close the generator explicitly with generator.close(), or use contextlib.closing().

Practical Example: A Stream with Safety Limits

This function streams an answer with three ways to stop it early, a character limit, a time limit, and Ctrl+C, and always closes the stream and reports why it stopped.

"""safe_stream.py: stream with character, time, and user-interrupt limits."""

import time
from dataclasses import dataclass

from anthropic import Anthropic

client = Anthropic()


@dataclass
class StreamOutcome:
    text: str
    reason: str        # "complete", "char_limit", "time_limit", "user_stopped"
    seconds: float


def stream_with_limits(question: str, max_chars: int = 2000, max_seconds: float = 30.0) -> StreamOutcome:
    start = time.perf_counter()
    chunks: list[str] = []
    length = 0
    reason = "complete"

    try:
        with client.messages.stream(
            model="claude-opus-5-5",
            max_tokens=4096,
            messages=[{"role": "user", "content": question}],
        ) as stream:
            for text in stream.text_stream:
                chunks.append(text)
                length += len(text)
                print(text, end="", flush=True)
                if length >= max_chars:
                    reason = "char_limit"
                    break
                if time.perf_counter() - start >= max_seconds:
                    reason = "time_limit"
                    break
    except KeyboardInterrupt:
        reason = "user_stopped"

    seconds = time.perf_counter() - start
    print(f"\n[{reason} after {seconds:.1f}s, {length} characters]")
    return StreamOutcome("".join(chunks), reason, seconds)


if __name__ == "__main__":
    outcome = stream_with_limits("List 40 practical Python tips, one paragraph each.", max_chars=600)
    if outcome.reason != "complete":
        print("Note: this answer is partial and should not be stored as complete.")

Example output:

1. Use virtual environments for every project, so dependencies never conflict...
2. Prefer f-strings for formatting...
[char_limit after 3.2s, 612 characters]
Note: this answer is partial and should not be stored as complete.

How It Works

1. The with block guarantees the stream is closed in every case: normal completion, break on a limit, or KeyboardInterrupt.
2. break on a limit leaves the loop, and the with block ends immediately after, closing the connection. get_final_message() is deliberately not called, because it would keep reading the rest of the response.
3. The time limit is checked on every chunk. It stops long-running answers even if the character limit has not been reached. It is checked only when a chunk arrives, so for a hard limit that also covers long pauses between chunks, use the async approach with asyncio.timeout().
4. except KeyboardInterrupt turns Ctrl+C into a recorded outcome instead of a crash. The with block closes the stream before the exception reaches the handler.
5. StreamOutcome.reason tells the caller exactly why the stream ended, so partial answers are never mistaken for complete ones. The message length reached 612 characters because the last chunk was added before the limit was checked.

Common Mistakes

Streams Without with or finally

What happens: an exception or early return leaves the connection open.

Fix: always use with/async with, or try/finally with close().

Calling get_final_message() After Deciding to Stop

What happens: the SDK keeps reading until the response completes, so nothing is saved.

Fix: leave the with block instead, and treat the collected text as partial.

Ignoring Disconnected Web Clients

What happens: the server keeps generating answers for users who have left, which wastes tokens and server capacity.

Fix: check await request.is_disconnected() in the streaming loop.

Treating Stopped Streams as Complete

What happens: partial answers are stored or displayed as finished.

Fix: record why the stream ended, and mark partial content.

Swallowing Cancellation in Async Code

What happens: catching asyncio.CancelledError and continuing prevents tasks from shutting down correctly.

Fix: let cancellation propagate, or re-raise it after cleanup. Handle TimeoutError from asyncio.timeout() at the level that set the time limit.

Best Practices

• Use with or async with for every stream.
• Stop early by leaving the block, not by abandoning the stream object.
• Do not call get_final_message() on a stream you are stopping.
• Add limits (time, size, user stop) for open-ended requests.
• Stop streams when web clients disconnect.
• Use try/finally when a stream must be passed around without a with block.
• Record and report why each stream ended.

Exercise

1. Run safe_stream.py with max_seconds=2.0 and a long question. Which limit is reached first?
2. Write an async version of stream_with_limits() that uses asyncio.timeout() for the time limit.
3. Use the stream_text() generator from this lesson with contextlib.closing(), so the stream is closed at a predictable moment when the loop breaks.
4. Explain why the time limit in safe_stream.py may not stop a stream that pauses for a long time between chunks, and how the async version avoids this.

Step 3:

from contextlib import closing

with closing(stream_text("List 50 Python tips.")) as chunks:
    for number, chunk in enumerate(chunks):
        print(chunk, end="")
        if number == 20:
            break
# The generator, and the stream inside it, are closed here

closing() calls the generator's close() method when the with block ends. Closing the generator raises GeneratorExit inside it, which runs the inner with block's cleanup and closes the stream immediately, rather than whenever the generator is garbage-collected.

Step 4: the sync version checks the elapsed time only when a chunk arrives. If no chunk arrives for a long time, the check does not run. asyncio.timeout() works differently: it schedules a cancellation at the deadline, which interrupts the waiting code, even in the middle of a pause, so the time limit is enforced exactly.