0Pricing
AI Agents · Lesson

Streaming Output in CLI Agents

Printing streamed tokens character-by-character in terminal interfaces.

Streaming Output in CLI Agents is a free AI Agents lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Streaming Matters for CLI Agents

Without streaming, your CLI agent prints nothing until the full LLM response is ready — this can take 5-30 seconds. Users stare at a blank terminal wondering if the program crashed.

With streaming, tokens appear as they are generated, providing immediate feedback and a much better experience.

Enabling Streaming in the OpenAI SDK

Pass stream=True to chat.completions.create(). The call returns a generator instead of a complete response object. Iterate over it to process chunks as they arrive.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

stream = client.chat.completions.create(
    model='gpt-4o-mini',
    messages=[{'role': 'user', 'content': 'Explain Python generators in 3 sentences.'}],
    stream=True  # <-- enable streaming
)

# Each chunk arrives as it is generated
for chunk in stream:
    delta = chunk.choices[0].delta
    if delta.content:
        print(delta.content, end='', flush=True)

print()  # newline after the response is complete

print() vs sys.stdout.write()

When streaming, use print(text, end='', flush=True) or sys.stdout.write(text) followed by sys.stdout.flush(). Without flush=True, Python may buffer output and print it all at once — defeating the purpose of streaming.

import sys
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_to_terminal(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    full_response = ''
    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_response += token

        # Option 1: print with flush
        print(token, end='', flush=True)

        # Option 2: sys.stdout.write + flush
        # sys.stdout.write(token)
        # sys.stdout.flush()

    print()  # final newline
    return full_response

Collecting the Full Response While Streaming

You often need the full response text after streaming completes — for storage, further processing, or display. Accumulate tokens into a string as you print them.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_and_collect(messages: list) -> str:
    full_text = ''

    print('Agent: ', end='', flush=True)

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_text += token
        print(token, end='', flush=True)

    print()  # newline
    return full_text

# The return value contains the complete response for storage
# response_text = stream_and_collect(history)
# history.append({'role': 'assistant', 'content': response_text})

Async Streaming with AsyncOpenAI

For async agent architectures, use AsyncOpenAI and async for to iterate over streaming chunks without blocking the event loop.

import asyncio
import openai

async def async_stream_agent(query: str) -> str:
    client = openai.AsyncOpenAI(api_key='YOUR_API_KEY')

    stream = await client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': query}],
        stream=True
    )

    full_text = ''
    print('Agent: ', end='', flush=True)

    async for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        full_text += token
        print(token, end='', flush=True)

    print()
    return full_text

# asyncio.run(async_stream_agent('What is asyncio?'))

Detecting Stream End with finish_reason

The last chunk in a stream has a non-null finish_reason. Check it to know why the stream ended: 'stop' = normal completion, 'length' = truncated, 'tool_calls' = function call needed.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_finish_detection(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    finish_reason = None
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end='', flush=True)
        if chunk.choices[0].finish_reason:
            finish_reason = chunk.choices[0].finish_reason

    print()

    if finish_reason == 'length':
        print('[WARNING: Response was truncated. Try increasing max_tokens.]')
    elif finish_reason == 'stop':
        pass  # normal completion

    return finish_reason

ANSI Colors in Terminal Output

ANSI escape codes add color to terminal output. Use them to visually distinguish the agent prefix, user input prompt, and warnings. The colorama library provides cross-platform support including Windows.

# pip install colorama
from colorama import Fore, Style, init
init(autoreset=True)  # reset color after each print

def print_colored_stream(messages: list, client):
    # Print agent prefix in cyan
    print(Fore.CYAN + 'Agent: ' + Style.RESET_ALL, end='', flush=True)

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True
    )

    for chunk in stream:
        token = chunk.choices[0].delta.content or ''
        print(token, end='', flush=True)

    print()

# Also useful:
# print(Fore.GREEN + 'Success!') — green
# print(Fore.RED + 'Error!') — red
# print(Fore.YELLOW + 'Warning') — yellow

Rich Library for Richer Terminal Output

The rich library provides markdown rendering, syntax-highlighted code blocks, tables, and spinners in the terminal. It pairs well with streamed agent output.

# pip install rich
from rich.console import Console
from rich.live import Live
from rich.markdown import Markdown

console = Console()

def stream_with_rich(messages: list, client):
    full_text = ''

    with Live(console=console, refresh_per_second=10) as live:
        stream = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=messages,
            stream=True
        )

        for chunk in stream:
            token = chunk.choices[0].delta.content or ''
            full_text += token
            # Render accumulated text as Markdown in real time
            live.update(Markdown(full_text))

    return full_text

Streaming with Tool Calls

When streaming is combined with function calling, the tool_calls field is also streamed in pieces. Accumulate the JSON string across chunks before parsing it.

import json
import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_tools(messages: list, tools: list) -> dict:
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        tools=tools,
        stream=True
    )

    tool_call_chunks = {}
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.tool_calls:
            for tc in delta.tool_calls:
                idx = tc.index
                if idx not in tool_call_chunks:
                    tool_call_chunks[idx] = {'name': '', 'args': ''}
                if tc.function.name:
                    tool_call_chunks[idx]['name'] += tc.function.name
                if tc.function.arguments:
                    tool_call_chunks[idx]['args'] += tc.function.arguments

    # Parse accumulated tool calls
    return {v['name']: json.loads(v['args']) for v in tool_call_chunks.values()}

Token Counter Display

Show a live token counter while streaming to help users monitor usage and understand costs. The OpenAI stream includes usage data in the final chunk when stream_options={'include_usage': True} is set.

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def stream_with_token_count(messages: list):
    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=messages,
        stream=True,
        stream_options={'include_usage': True}
    )

    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            print(delta.content, end='', flush=True)

        # Last chunk includes usage
        if chunk.usage:
            print(f'\n[Tokens: prompt={chunk.usage.prompt_tokens}, '
                  f'completion={chunk.usage.completion_tokens}, '
                  f'total={chunk.usage.total_tokens}]')

Streaming Best Practices

Summary of streaming output best practices for CLI agents:

  • Always use flush=True or sys.stdout.flush() to prevent buffering
  • Accumulate tokens into a string for storage after streaming
  • Check finish_reason to detect truncation
  • Use ANSI colors or rich for visual clarity
  • Handle tool call streaming by accumulating JSON argument chunks

Knowledge Check: Streaming Output

Test your understanding of streaming output in CLI agents.

Recap: Streaming Output in CLI Agents

You can now build streaming CLI agents that feel responsive and modern:

  • Pass stream=True to enable streaming from the OpenAI SDK
  • Use print(token, end='', flush=True) to display tokens immediately
  • Accumulate tokens into a string for post-stream processing
  • Check finish_reason on the last chunk to detect truncation
  • Use async for with AsyncOpenAI for async agents
  • Add ANSI colors or rich for a polished terminal experience

Frequently asked questions

Is the “Streaming Output in CLI Agents” lesson free?

Yes — the full text of “Streaming Output in CLI Agents” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.

What will I learn in “Streaming Output in CLI Agents”?

Printing streamed tokens character-by-character in terminal interfaces. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Agents?

No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Streaming Output in CLI Agents” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Agents lesson?

Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. Building Command-Line Agent Interfaces
  2. Interactive REPL-Style Agents
  3. Argument Parsing and Help Text
  4. Streaming Output in CLI Agents
← Back to AI Agents