Skip to main content
For the complete documentation index for agents and LLMs, see llms.txt.

Chat Stream

POST 
/api/v1/workspaces/:workspace_name/pipelines/:pipeline_name/chat-stream

Run a chat query against a pipeline, streaming the response as Server-Sent Events, using prior turns from a search session as chat history.

Chat pipelines are based on the chat template that uses a search session to include conversation history in the chat. You can specify how many search history items (query and answer) from a given search session you want to display in the chat.

You'll need a search session ID to group queries into a conversation. If omitted, a new session is created automatically and returned in the X-Search-Session-Id response header for reuse in subsequent requests. Use the search session endpoints for pipelines to manage search sessions.

Example request:

{
    "query": "How does streaming work with Haystack Enterprise Platform?",
    "search_session_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6"
}

Options:

  • search_session_id groups queries into a conversation, allowing them to use previous queries and answers as chat history. If omitted, a new session is created automatically. Its ID is returned in the X-Search-Session-Id response header for reuse in subsequent requests.
  • The full result is included as the last stream message if include_result=True.
  • Tool calls are streamed if include_tool_calls=True (default rendered, converted to markdown text deltas).
  • Tool call results are streamed if include_tool_call_results=True.
  • Reasoning output is streamed if include_reasoning=True.

Schema of each streamed event, where delta, result, tool_call_delta, tool_call_result, reasoning, ping and error are mutually exclusive:

{
    "query_id": UUID,
    "type": Literal["delta", "result", "error", "tool_call_delta", "tool_call_result", "reasoning", "ping"],
    "delta": Optional[StreamDelta],
    "result": Optional[DeepsetCloudQueryResponse],
    "error": Optional[str],
    "tool_call_delta": Optional[ToolCallDelta],
    "tool_call_result": Optional[ToolCallResult],
    "reasoning": Optional[ReasoningDelta],
    "index": Optional[int],
    "start": Optional[bool],
    "finish_reason": Optional[str],
}

StreamDelta format:

{
    "text": str,
    "meta": Optional[dict[str, Any]],
}

ToolCallDelta format:

{
    "index": int,
    "tool_name": Optional[str],
    "id": Optional[str],
    "arguments": Optional[str],
}

ToolCallResult format:

{
    "result": str,
    "origin": {
        "tool_name": str,
        "arguments": dict[str, Any],
        "id": Optional[str],
    },
    "error": bool,
}

ReasoningDelta format:

{
    "reasoning_text": str,
    "extra": dict[str, Any],
}

Example code to consume the stream in Python:

import httpx
import json
from httpx_sse import EventSource
import asyncio

TOKEN = "MY_TOKEN"
PIPELINE_URL = "https://api.cloud.deepset.ai/api/v1/workspaces/MY_WORKSPACE/pipelines/MY_PIPELINE"


async def main():
    query = {
        "query": "How does streaming work with deepset?",
        "search_session_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",  # optional: omit it to have deepset create one
        "include_tool_calls": True,
        "include_tool_call_results": True,
    }
    headers = {
        "Authorization": f"Bearer {TOKEN}"
    }
    async with httpx.AsyncClient(base_url=PIPELINE_URL, headers=headers, timeout=httpx.Timeout(300.0)) as client:
        async with client.stream("POST", "/chat-stream", json=query) as response:
            # Check if the response is successful
            if response.status_code != 200:
                await response.aread()
                print(f"An error occurred with status code: {response.status_code}")
                print(response.json()["errors"][0])
                return

            # Optionally save this to pass back as search_session_id on the next turn of the conversation.
            search_session_id = response.headers["X-Search-Session-Id"]

            event_source = EventSource(response)
            # Stream the response
            async for event in event_source.aiter_sse():
                event_data = json.loads(event.data)
                chunk_type = event_data["type"]
                # Check the type of the chunk and print the data accordingly
                match chunk_type:
                    # Delta chunk contains the next text chunk of the answer
                    case "delta":
                        delta = event_data["delta"]
                        if event_data.get("start"):
                            print(f"\n\nAnswer: ", flush=True, end="")
                        token: str = delta["text"]
                        print(token, flush=True, end="\n" if event_data.get("finish_reason") else "")
                    # Result chunk contains the final pipeline result
                    case "result":
                        print("\n\nPipeline result: ")
                        print(json.dumps(event_data["result"]))
                    # Error chunk contains the error message
                    case "error":
                        print("\n\nAn error occurred while streaming:")
                        print(event_data["error"])
                    case "tool_call_delta":
                        tool_call_delta = event_data["tool_call_delta"]
                        if tool_call_delta["tool_name"]:
                            tool_id = tool_call_delta["id"]
                            tool_name = tool_call_delta["tool_name"]
                            print(f"\n\nTool call {tool_id} started {tool_name} with arguments: ")
                        elif tool_call_delta["arguments"]:
                            print(tool_call_delta["arguments"], flush=True, end="")
                    case "tool_call_result":
                        tool_call_result = event_data["tool_call_result"]
                        tool_id = tool_call_result["origin"]["id"]
                        tool_name = tool_call_result["origin"]["tool_name"]
                        if tool_call_result["error"]:
                            print(f"\n\nTool call {tool_name} with id {tool_id} failed.")
                        else:
                            print(f"\n\nTool call {tool_name} with id {tool_id} result:")
                            print(tool_call_result["result"])
                    case "reasoning":
                        reasoning = event_data["reasoning"]
                        if event_data.get("start"):
                            print(f"\n\nReasoning: ")
                        print(f"{reasoning['reasoning_text']}", flush=True, end="")


asyncio.run(main())

Request​

Responses​

Returns the search results.