Implementation:Cohere ai Cohere python SSE EventSource

Field	Value
Source Repo	Cohere Python SDK
Domains	Streaming, HTTP, Event_Driven_Architecture
Last Updated	2026-02-15 14:00 GMT

Overview

Concrete SSE parsing infrastructure for decoding streaming HTTP responses into typed events.

Description

The SSE infrastructure consists of three classes:

• EventSource: Wraps an httpx.Response and provides iter_sse()/aiter_sse() iterators
• SSEDecoder: Stateful line-by-line decoder that accumulates event fields and emits ServerSentEvent objects on blank line boundaries
• ServerSentEvent: Frozen dataclass with event, data, id, retry fields and a json() method

EventSource handles charset detection from Content-Type headers and validates the text/event-stream content type.

Usage

Used internally by the streaming client. The EventSource.iter_sse() method yields ServerSentEvent objects which are then parsed into typed V2ChatStreamResponse models by the SDK.

Code Reference

• Source Location: Repository cohere-ai/cohere-python https://github.com/cohere-ai/cohere-python
  • File src/cohere/core/http_sse/_api.py, Lines L13-87 (EventSource)
  • File src/cohere/core/http_sse/_decoders.py, Lines L8-61 (SSEDecoder)
  • File src/cohere/core/http_sse/_models.py, Lines L1-17 (ServerSentEvent)

Signature:

class EventSource:
    def __init__(self, response: httpx.Response) -> None: ...
    def iter_sse(self) -> Iterator[ServerSentEvent]: ...
    async def aiter_sse(self) -> AsyncGenerator[ServerSentEvent, None]: ...

class SSEDecoder:
    def decode(self, line: str) -> Optional[ServerSentEvent]: ...

@dataclass(frozen=True)
class ServerSentEvent:
    event: str = "message"
    data: str = ""
    id: str = ""
    retry: Optional[int] = None
    def json(self) -> Any: ...

Import:

from cohere.core.http_sse import EventSource, SSEDecoder, ServerSentEvent  # internal; not typically imported by users

I/O Contract

Inputs

Component	Input	Description
EventSource	httpx.Response	HTTP response with content-type text/event-stream
SSEDecoder.decode	str	Individual SSE text line

Outputs

Component	Output	Description
EventSource.iter_sse()	Iterator[ServerSentEvent]	Synchronous iterator of parsed SSE events
EventSource.aiter_sse()	AsyncGenerator[ServerSentEvent, None]	Asynchronous iterator of parsed SSE events
SSEDecoder.decode	Optional[ServerSentEvent]	Returns a ServerSentEvent on blank line boundary, None otherwise
ServerSentEvent.json()	Any	Parses the data field as JSON and returns the result

ServerSentEvent fields:

Field	Type	Default	Description
event	str	"message"	Event type name
data	str	""	Event data payload
id	str	""	Event ID
retry	Optional[int]	None	Reconnection time in milliseconds

Usage Examples

# Internal SDK usage (not typically called directly)
from cohere.core.http_sse._api import EventSource

# EventSource wraps an httpx streaming response
with httpx.Client() as http_client:
    with http_client.stream("POST", url, headers=headers, json=body) as response:
        source = EventSource(response)
        for sse in source.iter_sse():
            event_data = sse.json()
            print(f"Event: {sse.event}, Data: {event_data}")

Related Pages

• Principle:Cohere_ai_Cohere_python_Server_Sent_Events_Decoding
• Environment:Cohere_ai_Cohere_python_Python_SDK_Runtime
• Heuristic:Cohere_ai_Cohere_python_Pydantic_V1_V2_Compatibility