Implementation:Cohere ai Cohere python GenerateStreamedResponse Union

Knowledge Sources	Cohere Python SDK
Domains	SDK, Text Generation, Streaming
Last Updated	2026-02-15 14:00 GMT

Overview

GenerateStreamedResponse is a discriminated union type representing the different event types emitted during a streaming text generation request from the Cohere Generate API.

Description

The GenerateStreamedResponse type is a discriminated union (tagged union) composed of three variant classes, each representing a different event in the streaming generate response lifecycle:

• TextGenerationGenerateStreamedResponse: Emitted for each chunk of generated text during streaming. Contains the generated text token, an optional index (for multi-generation requests), and an is_finished flag. Discriminator value: "text-generation".

• StreamEndGenerateStreamedResponse: Emitted when the stream ends successfully. Contains an is_finished flag, an optional finish_reason (such as "COMPLETE", "STOP_SEQUENCE", "MAX_TOKENS", etc.), and a response object of type GenerateStreamEndResponse containing the final aggregated result. Discriminator value: "stream-end".

• StreamErrorGenerateStreamedResponse: Emitted when an error occurs during streaming. Contains an is_finished flag, a finish_reason, an optional index, and an err string with the error message. Discriminator value: "stream-error".

The union uses the event_type field as the discriminant, annotated via UnionMetadata(discriminant="event_type").

Usage

Use GenerateStreamedResponse when consuming streamed responses from the Cohere Generate API with stream=True. Each event in the stream is deserialized as one of the three variant types, allowing real-time token-by-token text display and proper handling of stream completion or errors.

Code Reference

Source Location

• Repository: Cohere Python SDK
• File: src/cohere/types/generate_streamed_response.py

Signature

class TextGenerationGenerateStreamedResponse(UncheckedBaseModel):
    event_type: typing.Literal["text-generation"] = "text-generation"
    text: str
    index: typing.Optional[int] = None
    is_finished: bool

class StreamEndGenerateStreamedResponse(UncheckedBaseModel):
    event_type: typing.Literal["stream-end"] = "stream-end"
    is_finished: bool
    finish_reason: typing.Optional[FinishReason] = None
    response: GenerateStreamEndResponse

class StreamErrorGenerateStreamedResponse(UncheckedBaseModel):
    event_type: typing.Literal["stream-error"] = "stream-error"
    index: typing.Optional[int] = None
    is_finished: bool
    finish_reason: FinishReason
    err: str

GenerateStreamedResponse = typing_extensions.Annotated[
    typing.Union[
        TextGenerationGenerateStreamedResponse,
        StreamEndGenerateStreamedResponse,
        StreamErrorGenerateStreamedResponse,
    ],
    UnionMetadata(discriminant="event_type"),
]

Import

from cohere.types import GenerateStreamedResponse
from cohere.types.generate_streamed_response import (
    TextGenerationGenerateStreamedResponse,
    StreamEndGenerateStreamedResponse,
    StreamErrorGenerateStreamedResponse,
)

I/O Contract

Union Variants

Variant	Discriminator Value	Description
TextGenerationGenerateStreamedResponse	"text-generation"	A chunk of generated text tokens during streaming
StreamEndGenerateStreamedResponse	"stream-end"	Final event indicating the stream has ended successfully
StreamErrorGenerateStreamedResponse	"stream-error"	Error event indicating a failure during streaming

TextGenerationGenerateStreamedResponse Fields

Field	Type	Required	Default	Description
event_type	Literal["text-generation"]	Yes	"text-generation"	Discriminator identifying this as a text generation event
text	str	Yes	--	The generated text token
index	Optional[int]	No	None	Index of the generation (for multi-generation requests)
is_finished	bool	Yes	--	Whether this generation is finished

StreamEndGenerateStreamedResponse Fields

Field	Type	Required	Default	Description
event_type	Literal["stream-end"]	Yes	"stream-end"	Discriminator identifying this as a stream end event
is_finished	bool	Yes	--	Whether the stream is finished
finish_reason	Optional[FinishReason]	No	None	Reason the generation finished (e.g., "COMPLETE", "MAX_TOKENS", "STOP_SEQUENCE")
response	GenerateStreamEndResponse	Yes	--	The final aggregated generation response

StreamErrorGenerateStreamedResponse Fields

Field	Type	Required	Default	Description
event_type	Literal["stream-error"]	Yes	"stream-error"	Discriminator identifying this as a stream error event
index	Optional[int]	No	None	Index of the generation that errored
is_finished	bool	Yes	--	Whether the stream is finished
finish_reason	FinishReason	Yes	--	Reason the generation finished (e.g., "ERROR", "ERROR_TOXIC")
err	str	Yes	--	The error message

FinishReason Values

Value	Description
"COMPLETE"	Generation completed normally
"STOP_SEQUENCE"	A stop sequence was encountered
"ERROR"	A general error occurred
"ERROR_TOXIC"	Content was flagged as toxic
"ERROR_LIMIT"	A limit was reached
"USER_CANCEL"	The user cancelled the generation
"MAX_TOKENS"	The maximum token limit was reached
"TIMEOUT"	The generation timed out

Usage Examples

Consuming a Streamed Generate Response

import cohere
from cohere.types.generate_streamed_response import (
    TextGenerationGenerateStreamedResponse,
    StreamEndGenerateStreamedResponse,
    StreamErrorGenerateStreamedResponse,
)

co = cohere.Client()

stream = co.generate_stream(
    prompt="Write a short poem about the ocean:",
    model="command",
    max_tokens=200,
)

generated_text = ""
for event in stream:
    if isinstance(event, TextGenerationGenerateStreamedResponse):
        # Accumulate text tokens as they arrive
        print(event.text, end="", flush=True)
        generated_text += event.text
    elif isinstance(event, StreamEndGenerateStreamedResponse):
        print()  # newline after streaming
        print(f"Finish reason: {event.finish_reason}")
        print(f"Final response ID: {event.response.id}")
    elif isinstance(event, StreamErrorGenerateStreamedResponse):
        print(f"Stream error: {event.err}")
        print(f"Finish reason: {event.finish_reason}")

Handling Multiple Generations

import cohere
from cohere.types.generate_streamed_response import (
    TextGenerationGenerateStreamedResponse,
    StreamEndGenerateStreamedResponse,
)

co = cohere.Client()

stream = co.generate_stream(
    prompt="Suggest a name for a cat:",
    model="command",
    num_generations=3,
    max_tokens=50,
)

generations = {}
for event in stream:
    if isinstance(event, TextGenerationGenerateStreamedResponse):
        idx = event.index or 0
        generations.setdefault(idx, "")
        generations[idx] += event.text
    elif isinstance(event, StreamEndGenerateStreamedResponse):
        print("Stream complete!")

for idx, text in sorted(generations.items()):
    print(f"Generation {idx}: {text.strip()}")

Related Pages

• Environment:Cohere_ai_Cohere_python_Python_SDK_Runtime