C.W.K.
Stream
Lesson 02 of 04 · published

Python Streaming

~12 min · python, streaming, async-iterator

Level 0Spark
0 XP0/35 lessons0/10 achievements
0/140 XP to next level140 XP to go0% complete

generate_content_stream — same args, iterator return

The streaming methods sit alongside the non-streaming ones and take the same arguments. The only difference is the return value: instead of a single GenerateContentResponse, you get an iterator (sync) or async iterator (async) of partial responses.

Three patterns to remember

  • Sync stream: for chunk in client.models.generate_content_stream(...)
  • Async stream: async for chunk in await client.aio.models.generate_content_stream(...) (note the await on the call itself before the async for)
  • Streaming chat: chat.send_message_stream(...) / chat.aio.send_message_stream(...)

Concatenate as you go, grab usage at the end

Each chunk's chunk.text is the partial text for that slice. Concatenate them into the full reply. usage_metadata appears only on the final chunk — capture it lazily as you iterate.

Multimodal streams the same way

You can pass an image, an uploaded file, or any of the same multi-part contents you'd use in non-streaming. The stream begins as soon as the model produces its first text token.

Code

Sync streaming·python
from google import genai

client = genai.Client()

for chunk in client.models.generate_content_stream(
    model='gemini-2.5-flash',
    contents='Tell me a 200-word story about a coffee scale.',
):
    if chunk.text:
        print(chunk.text, end='', flush=True)
print()  # final newline
Async streaming with usage capture·python
import asyncio
from google import genai

client = genai.Client()

async def stream_with_usage():
    full_text = []
    final_usage = None

    async for chunk in await client.aio.models.generate_content_stream(
        model='gemini-2.5-flash',
        contents='Write a 300-word story about an octopus learning JavaScript.',
    ):
        if chunk.text:
            print(chunk.text, end='', flush=True)
            full_text.append(chunk.text)
        if chunk.usage_metadata:
            final_usage = chunk.usage_metadata

    print(f'\n\n[total tokens: {final_usage.total_token_count}]')
    return ''.join(full_text)

asyncio.run(stream_with_usage())
Streaming chat·python
chat = client.chats.create(model='gemini-2.5-flash')

for chunk in chat.send_message_stream('Tell me a haiku.'):
    if chunk.text:
        print(chunk.text, end='', flush=True)
print()

for chunk in chat.send_message_stream('And another.'):
    if chunk.text:
        print(chunk.text, end='', flush=True)
print()
Streaming with image input·python
from google.genai import types

for chunk in client.models.generate_content_stream(
    model='gemini-2.5-flash',
    contents=[
        'Describe what's happening in this image:',
        types.Part.from_uri(
            file_uri='gs://my-bucket/cafe.jpg',
            mime_type='image/jpeg',
        ),
    ],
):
    if chunk.text:
        print(chunk.text, end='', flush=True)

External links

Exercise

Build a CLI that takes a prompt as argv, streams Flash's response to stdout, and at the end prints: total tokens used, wall-clock time elapsed, and tokens-per-second. Then add a flag --sync that uses the sync streaming surface instead of async. Compare the perceived smoothness of the two.

Progress

Progress is local-only — sign in to sync across devices.
Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)
Sign inPlease sign in to comment.

No comments yet — be the first.