Streaming Gemini Interactions: SSE Events, Tools, Thinking, and Agents

Introduction

Streaming is essential for building responsive AI applications. Instead of waiting for a model to finish generating an entire response, you can stream output incrementally and show it to users as it arrives. The Gemini Interactions API supports this natively.

When you create an interaction with stream: true, the API returns the response using server-sent events (SSE). Each event has a named event_type and associated JSON data. To make this approach elegant, the Interactions API uses a symmetric, step-based streaming model where all content — text, tool calls, thinking — flows through the same consistent event structure.

In this post we’ll break down the event flow, walk through streaming with tools, function calling, thinking summaries, agents, and even image generation, then cover how to handle unknown events gracefully.

Basic streaming

Setting stream: true is all it takes to start receiving events incrementally.

Python

from google import genai

client = genai.Client()

stream = client.interactions.create(
    model="gemini-3.6-flash",
    input="Count from 1 to 25.",
    stream=True,
)
for event in stream:
    if event.event_type == "step.delta":
        if event.delta.type == "text":
            print(event.delta.text, end="", flush=True)

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI({});

const stream = await client.interactions.create({
    model: "gemini-3.6-flash",
    input: "Count from 1 to 25.",
    stream: true,
});
for await (const event of stream) {
    if (event.event_type === "step.delta") {
        if (event.delta.type === "text") {
            process.stdout.write(event.delta.text);
        }
    }
}

REST

curl -X POST "https://generativelanguage.googleapis.com/v1beta/interactions" \
  -H "x-goog-api-key: $GEMINI_API_KEY" \
  -H "Content-Type: application/json" \
  --no-buffer \
  -d '{
    "model": "gemini-3.6-flash",
    "input": "Count from 1 to 25.",
    "stream": true
  }'

The stream emits a predictable sequence of events: interaction.created, a series of step cycles (step.start → step.delta(s) → step.stop), an interaction.completed with final usage, and a terminal done.

Understanding the event flow

Each stream follows this flow:

1. interaction.created: The interaction is created, with metadata (ID, model, status).
2. A series of steps, each consisting of:
   • A step.start event indicating the step type (e.g., model_output, thought, function_call).
   • One or more step.delta events with incremental data for that step.
   • A step.stop event marking the step complete.
3. An interaction.completed event with final usage statistics.

When you set stream: false, the API returns a single interaction object with a steps array — each element is the fully assembled version of one step.start → step.delta(s) → step.stop cycle.

Event types at a glance

interaction.created — Sent first, contains the interaction ID, model, and initial status.

interaction.status_update — Signals an interaction-level status transition, and may appear between steps.

step.start — Marks the beginning of a step. Contains the step type and index. The step type determines which delta types to expect:

Step Type	Expected Delta Types	Description
model_output	text, image, audio	The model’s final response content.
thought	thought_signature, thought_summary	Chain-of-thought reasoning. summary is only present when thinking_summaries is enabled.
function_call	arguments_delta	A request for the client to execute a function. Sets status to requires_action.
Server-side tools	Varies by tool	Tools executed by the API (e.g., google_search_call, code_execution_call).

step.delta — Incremental data for the current step. The delta object has a type field that determines its shape. Common examples include text, image (base64-encoded), thought_summary, and arguments_delta (partial JSON for function-call arguments you must accumulate across deltas).

step.stop — Marks the end of a step, containing the step index. When using the Antigravity Agent, this event may also include usage and step_usage statistics.

interaction.completed — Sent when finished, with the final interaction object and usage statistics.

error — Sent when an error occurs, containing an error object with a message and code.

Streaming with tools

The Interactions API supports streaming with both client-side tools (function calling) and server-side tools (Google Search, Code Execution) in a single request. During streaming, tool invocations appear as typed steps in the event stream.

Function calling (multi-turn)

Function calling with streaming requires the client to handle a two-turn conversation:

1. Turn 1 (Function Request): Call interactions.create with stream: true and your tools. The API streams a function_call step. Accumulate the incremental argument JSON strings (arguments_delta) until the interaction completes with status requires_action.
2. Turn 2 (Sending Result): Call interactions.create again with previous_interaction_id (the first interaction’s ID) and a function_result block in the input array. This resumes the stream so the model can generate its final response.

Python

from google import genai

client = genai.Client()

weather_tool = {
    "type": "function",
    "name": "get_weather",
    "description": "Get the current weather in a given location",
    "parameters": {
        "type": "object",
        "properties": {
            "location": {
                "type": "string",
                "description": "The city and state, e.g. San Francisco, CA"
            }
        },
        "required": ["location"]
    }
}

# Turn 1: Request function call
stream = client.interactions.create(
    model="gemini-3.6-flash",
    tools=[weather_tool],
    input="What is the weather in Paris right now?",
    stream=True,
)

first_interaction_id = None
func_call_id = None
func_call_name = None
func_args_accumulated = ""

for event in stream:
    if event.event_type == "interaction.created":
        first_interaction_id = event.interaction.id
    elif event.event_type == "step.start":
        step = event.step
        if step.type == "function_call":
            func_call_id = step.id
            func_call_name = step.name
    elif event.event_type == "step.delta":
        if event.delta.type == "arguments_delta":
            func_args_accumulated += event.delta.arguments

# Turn 2: Execute tool and send the result back to resume stream
if func_call_id:
    # Execute weather_tool using accumulated arguments
    dummy_result = {
        "content": [{"type": "text", "text": '{"weather": "Sunny and 22°C"}'}]
    }

    stream2 = client.interactions.create(
        model="gemini-3.6-flash",
        previous_interaction_id=first_interaction_id,
        input=[{
            "type": "function_result",
            "name": func_call_name,
            "call_id": func_call_id,
            "result": dummy_result
        }],
        stream=True,
    )

    for event in stream2:
        if event.event_type == "step.delta":
            if event.delta.type == "text":
                print(event.delta.text, end="", flush=True)

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI({});

const weatherTool = {
    type: "function",
    name: "get_weather",
    description: "Get the current weather in a given location",
    parameters: {
        type: "object",
        properties: {
            location: {
                type: "string",
                description: "The city and state, e.g. San Francisco, CA"
            }
        },
        required: ["location"]
    }
};

// Turn 1: Request function call
const stream = await client.interactions.create({
    model: "gemini-3.6-flash",
    tools: [weatherTool],
    input: "What is the weather in Paris right now?",
    stream: true,
});

let firstInteractionId = null;
let funcCallId = null;
let funcCallName = null;
let funcArgsAccumulated = "";

for await (const event of stream) {
    if (event.event_type === "interaction.created") {
        firstInteractionId = event.interaction.id;
    } else if (event.event_type === "step.start") {
        const step = event.step;
        if (step.type === "function_call") {
            funcCallId = step.id;
            funcCallName = step.name;
        }
    } else if (event.event_type === "step.delta") {
        if (event.delta.type === "arguments_delta") {
            funcArgsAccumulated += event.delta.arguments;
        }
    }
}

// Turn 2: Execute tool and send the result back to resume stream
if (funcCallId && firstInteractionId && funcCallName) {
    const dummyResult = {
        content: [{ type: "text", text: '{"weather": "Sunny and 22°C"}' }]
    };

    const stream2 = await client.interactions.create({
        model: "gemini-3.6-flash",
        previous_interaction_id: firstInteractionId,
        input: [{
            type: "function_result",
            name: funcCallName,
            call_id: funcCallId,
            result: dummyResult
        }],
        stream: true,
    });

    for await (const event of stream2) {
        if (event.event_type === "step.delta") {
            if (event.delta.type === "text") {
                process.stdout.write(event.delta.text);
            }
        }
    }
}

Streaming with multiple tools

You can combine a function tool and google_search in one request. The event stream will interleave google_search_call/google_search_result steps and function_call steps.

Python

from google import genai

client = genai.Client()

tools = [
    {"type": "google_search"},
    {
        "type": "function",
        "name": "get_weather",
        "description": "Get the current weather in a given location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "The city and state, e.g. San Francisco, CA"
                }
            },
            "required": ["location"]
        }
    }
]

stream = client.interactions.create(
    model="gemini-3.6-flash",
    tools=tools,
    input="Search what is the largest mountain in Europe and what the weather is there right now?",
    stream=True,
)
for event in stream:
    if event.event_type == "step.start":
        step = event.step
        print(f"\n--- Step {event.index}: {step.type} ---")
        if step.type == "google_search_call":
            print(f"  Search ID: {step.id}")
        elif step.type == "google_search_result":
            print(f"  Result for: {step.call_id}")
        elif step.type == "function_call":
            print(f"  Function: {step.name}({step.arguments})")
    elif event.event_type == "step.delta":
        if event.delta.type == "text":
            print(event.delta.text, end="", flush=True)
        elif event.delta.type == "google_search_call":
            print(f"  Queries: {event.delta.arguments}")
        elif event.delta.type == "arguments_delta":
            print(f"  Args chunk: {event.delta.arguments}", end="", flush=True)
    elif event.event_type == "interaction.completed":
        print(f"\n\nStatus: {event.interaction.status}")
        if event.interaction.status == "requires_action":
            print("Action required: provide function call results to continue.")

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI({});

const tools = [
    { type: "google_search" },
    {
        type: "function",
        name: "get_weather",
        description: "Get the current weather in a given location",
        parameters: {
            type: "object",
            properties: {
                location: {
                    type: "string",
                    description: "The city and state, e.g. San Francisco, CA"
                }
            },
            required: ["location"]
        }
    }
];

const stream = await client.interactions.create({
    model: "gemini-3.6-flash",
    tools: tools,
    input: "Search what is the largest mountain in Europe and what the weather is there right now?",
    stream: true,
});
for await (const event of stream) {
    if (event.event_type === "step.start") {
        const step = event.step;
        console.log(`\n--- Step ${event.index}: ${step.type} ---`);
        if (step.type === "google_search_call") {
            console.log(`  Search ID: ${step.id}`);
        } else if (step.type === "google_search_result") {
            console.log(`  Result for: ${step.call_id}`);
        } else if (step.type === "function_call") {
            console.log(`  Function: ${step.name}(${JSON.stringify(step.arguments)})`);
        }
    } else if (event.event_type === "step.delta") {
        if (event.delta.type === "text") {
            process.stdout.write(event.delta.text);
        } else if (event.delta.type === "google_search_call") {
            console.log(`  Queries: ${JSON.stringify(event.delta.arguments?.queries)}`);
        } else if (event.delta.type === "arguments_delta") {
            process.stdout.write(`  Args chunk: ${event.delta.arguments}`);
        }
    } else if (event.event_type === "interaction.completed") {
        console.log(`\n\nStatus: ${event.interaction.status}`);
        if (event.interaction.status === "requires_action") {
            console.log("Action required: provide function call results to continue.");
        }
    }
}

Streaming with thinking

When the model uses thinking, you’ll receive thought steps with two distinct delta types: thought_summary (incremental text or image summary content) and thought_signature (an encrypted representation of the model’s internal reasoning, sent as the last delta before step.stop). If thinking_summaries is enabled, thought_summary deltas stream a summary of the model’s reasoning.

Python

from google import genai

client = genai.Client()

stream = client.interactions.create(
    model="gemini-3.6-flash",
    input="What is the greatest common divisor of 1071 and 462?",
    generation_config={
        "thinking_summaries": "auto"
    },
    stream=True,
)
for event in stream:
    if event.event_type == "step.start":
        print(f"\n--- Step: {event.step.type} ---")
    elif event.event_type == "step.delta":
        if event.delta.type == "thought_summary":
            if event.delta.content.type == "text":
                print(event.delta.content.text, end="", flush=True)
        elif event.delta.type == "text":
            print(event.delta.text, end="", flush=True)

On a math problem like computing the GCD of 1071 and 462, you’ll see a thought step streaming a human-readable summary of the Euclidean algorithm, followed by a model_output step with the final answer.

Streaming with agents

The Interactions API supports agents like Deep Research. Agents use background=True and return results asynchronously, but you can stream agent interactions to receive progress updates and intermediate steps as they happen.

Python

from google import genai

client = genai.Client()

stream = client.interactions.create(
    agent="deep-research-preview-04-2026",
    input="Research the latest advances in quantum computing.",
    stream=True,
    background=True,
    agent_config={
        "type": "deep-research",
        "thinking_summaries": "auto"
    }
)
for event in stream:
    if event.event_type == "step.start":
        print(f"\n--- Step: {event.step.type} ---")
    elif event.event_type == "step.delta":
        if event.delta.type == "text":
            print(event.delta.text, end="", flush=True)
        elif event.delta.type == "thought_summary":
            if event.delta.content.type == "text":
                print(event.delta.content.text, end="", flush=True)
    elif event.event_type == "interaction.completed":
        print(f"\n\nTotal Tokens: {event.interaction.usage.total_tokens}")

With Deep Research, you’ll stream the agent’s research plan as thought_summary steps, then the final model_output. The interaction.completed event reveals just how token-hungry deep research can be — in the reference example, over a million total tokens.

Streaming image generation

The Interactions API supports streaming multiple output modalities simultaneously. By requesting both text and image in the response_format, you can receive interleaved text and generated images in the same stream. This example uses gemini-3.1-flash-image (Nano Banana 2) to search and generate an illustrated story.

Python

from google import genai

client = genai.Client()

stream = client.interactions.create(
    model="gemini-3.1-flash-image",
    tools=[{"type": "google_search", "search_types": ["web_search", "image_search"]}],
    input="Search for the history of the Colosseum and write a short illustrated story about a gladiator named Marcus. Interleave text and generated images.",
    response_format=[
        {"type": "text"},
        {"type": "image"}
    ],
    stream=True,
)

for event in stream:
    if event.event_type == "step.delta":
        if event.delta.type == "text":
            print(event.delta.text, end="", flush=True)
        elif event.delta.type == "image":
            print(f"\n[Image chunk: {len(event.delta.data)} bytes]", end="", flush=True)

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI({});

const stream = await client.interactions.create({
    model: "gemini-3.1-flash-image",
    tools: [{ type: "google_search", search_types: ["web_search", "image_search"] }],
    input: "Search for the history of the Colosseum and write a short illustrated story about a gladiator named Marcus. Interleave text and generated images.",
    response_format: [
        { type: "text" },
        { type: "image" }
    ],
    stream: true,
});

for await (const event of stream) {
    if (event.event_type === "step.delta") {
        if (event.delta.type === "text") {
            process.stdout.write(event.delta.text);
        } else if (event.delta.type === "image") {
            console.log(`\n[Image chunk: ${event.delta.data.length} bytes]`);
        }
    }
}

Handling unknown events

In accordance with the API’s versioning policy, new event types and delta types may be added over time. Your code should handle unknown events gracefully — log and skip any events you don’t recognize rather than throwing an error. The default else in your handler (or simply not matching on unknown types) is enough to keep your application forward-compatible.

Summary

Streaming transforms the Gemini Interactions API from a request/response tool into a live, event-driven interface. The symmetric step-based model means text, tool calls, and thinking all flow through one consistent structure. By understanding the event flow — interaction.created, step cycles, interaction.completed — you can build responsive chat UIs, agent orchestration, multi-tool pipelines, and even interleaved text-plus-image generation, all while staying resilient to future API changes.

Further reading

• Learn more about the Interactions API.
• Explore Function calling with tools.
• Learn about Thinking for enhanced reasoning.
• Try the Deep Research Agent for long-running tasks.
• See the Interactions API reference for all event types and delta types.