feat: add conversation stream config builder

* add typed streaming callbacks and decorator-based builder for sse responses
* extend search parameters to accept conversation streaming configuration

Author: tharropoulos

src/typesense/types/document.py

@@ -586,6 +586,159 @@ class NLLanguageParameters(typing.TypedDict):
     nl_query_debug: typing.NotRequired[bool]
 
 
+class MessageChunk(typing.TypedDict):
+    """
+    A single chunk from a conversation stream response.
+
+    Attributes:
+      conversation_id (str): ID of the conversation.
+      message (str): Message content for this chunk.
+    """
+
+    conversation_id: str
+    message: str
+
+
+class StreamConfig(typing.Generic[TDoc], typing.TypedDict, total=False):
+    """
+    Configuration for streaming conversation search responses.
+
+    Attributes:
+      on_chunk: Callback invoked for each streamed chunk (conversation_id, message).
+      on_complete: Callback invoked when the stream completes with the full search response.
+      on_error: Callback invoked if an error occurs during streaming.
+    """
+
+    on_chunk: typing.Callable[["MessageChunk"], None]
+    on_complete: typing.Callable[["SearchResponse[TDoc]"], None]
+    on_error: typing.Callable[[BaseException], None]
+
+
+OnChunkCallback = typing.Callable[["MessageChunk"], None]
+OnCompleteCallback = typing.Callable[["SearchResponse[TDoc]"], None]
+OnErrorCallback = typing.Callable[[BaseException], None]
+
+_OnChunkT = typing.TypeVar("_OnChunkT", bound=OnChunkCallback)
+_OnCompleteT = typing.TypeVar("_OnCompleteT", bound=OnCompleteCallback)
+_OnErrorT = typing.TypeVar("_OnErrorT", bound=OnErrorCallback)
+
+
+class StreamConfigBuilder:
+    """
+    Builder for StreamConfig using decorators.
+
+    Example:
+        >>> stream = StreamConfigBuilder()
+        >>>
+        >>> @stream.on_chunk
+        ... def handle_chunk(chunk: MessageChunk) -> None:
+        ...     print(chunk["message"], end="", flush=True)
+        >>>
+        >>> @stream.on_complete
+        ... def handle_complete(response: dict) -> None:
+        ...     print(f"Done! Found {response.get('found', 0)}")
+        >>>
+        >>> response = client.collections["docs"].documents.search({
+        ...     "q": "query",
+        ...     "query_by": "content",
+        ...     "conversation_stream": True,
+        ...     "stream_config": stream,
+        ... })
+    """
+
+    def __init__(self) -> None:
+        """Initialize an empty StreamConfigBuilder."""
+        self._on_chunk: typing.Optional[OnChunkCallback] = None
+        self._on_complete: typing.Optional[OnCompleteCallback] = None
+        self._on_error: typing.Optional[OnErrorCallback] = None
+
+    def on_chunk(self, func: _OnChunkT) -> _OnChunkT:
+        """
+        Decorator to register an on_chunk callback.
+
+        Args:
+            func: Callback invoked for each streamed message chunk.
+
+        Returns:
+            The original function (unmodified).
+        """
+        self._on_chunk = func
+        return func
+
+    def on_complete(self, func: _OnCompleteT) -> _OnCompleteT:
+        """
+        Decorator to register an on_complete callback.
+
+        Args:
+            func: Callback invoked when streaming completes with the full response.
+
+        Returns:
+            The original function (unmodified).
+        """
+        self._on_complete = func
+        return func
+
+    def on_error(self, func: _OnErrorT) -> _OnErrorT:
+        """
+        Decorator to register an on_error callback.
+
+        Args:
+            func: Callback invoked if an error occurs during streaming.
+
+        Returns:
+            The original function (unmodified).
+        """
+        self._on_error = func
+        return func
+
+    def build(self) -> StreamConfig:
+        """
+        Build the StreamConfig dictionary.
+
+        Returns:
+            A StreamConfig with the registered callbacks.
+        """
+        config: StreamConfig = {}
+        if self._on_chunk is not None:
+            config["on_chunk"] = self._on_chunk
+        if self._on_complete is not None:
+            config["on_complete"] = self._on_complete
+        if self._on_error is not None:
+            config["on_error"] = self._on_error
+        return config
+
+    def get(
+        self, key: str, default: typing.Any = None
+    ) -> typing.Optional[typing.Callable[..., None]]:
+        """
+        Get a callback by key (for compatibility with dict-like access).
+
+        Args:
+            key: The callback name ('on_chunk', 'on_complete', or 'on_error').
+            default: Default value if the callback is not set.
+
+        Returns:
+            The callback function or the default value.
+        """
+        return self.build().get(key, default)
+
+
+class ConversationStreamParameters(typing.Generic[TDoc], typing.TypedDict):
+    """
+    Parameters for conversational search streaming.
+
+    Attributes:
+      conversation_stream (bool): When true, the search response is streamed (SSE).
+      stream_config: Callbacks for stream events. Not sent to the API.
+        Can be a StreamConfig dict or a StreamConfigBuilder instance.
+    """
+
+    conversation_stream: typing.NotRequired[bool]
+    stream_config: typing.NotRequired[
+        typing.Union[StreamConfig[TDoc], StreamConfigBuilder]
+    ]
+
+
 class SearchParameters(
     RequiredSearchParameters,
     QueryParameters,
@@ -598,6 +751,7 @@ class SearchParameters(
     TypoToleranceParameters,
     CachingParameters,
     NLLanguageParameters,
+    ConversationStreamParameters,
 ):
     """Parameters for searching documents."""