# Chat Stream

<Heading
  as={"h1"}
  className={"openapi__heading"}
  children={"Chat Stream"}
>
</Heading>

<MethodEndpoint
  method={"post"}
  path={"/api/v1/workspaces/{workspace_name}/pipelines/{pipeline_name}/chat-stream"}
  context={"endpoint"}
>
  
</MethodEndpoint>

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:
```json
{
    "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`.

Ping events are connection status, not answer text. `{"type":"ping"}` is a heartbeat, while
`{"type":"ping","phase":"cold_start"}` tells you that temporary infrastructure and pipeline preparation is
still running. You can ignore either form. After you receive a ping, later failures arrive as one terminal
`error` event over the HTTP 200 stream and are not followed by `done`.

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],
}
```

PingEvent format (unlike other events, a ping has no `query_id`):
```
{
    "type": "ping",
    "phase": 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:
```python
from httpx_sse import EventSource

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"])
                    # Ping chunks are connection status, not answer text, and can be ignored
                    case "ping":
                        continue
                    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())
```

<Heading
  id={"request"}
  as={"h2"}
  className={"openapi-tabs__heading"}
>
  <Translate id="theme.openapi.request.title">Request</Translate>
</Heading>

<ParamsDetails
  {...require("./chat-stream-api-v-1-workspaces-workspace-name-pipelines-pipeline-name-chat-stream-post.ParamsDetails.json")}
>
  
</ParamsDetails>

<RequestSchema
  {...require("./chat-stream-api-v-1-workspaces-workspace-name-pipelines-pipeline-name-chat-stream-post.RequestSchema.json")}
>
  
</RequestSchema>

<StatusCodes
  {...require("./chat-stream-api-v-1-workspaces-workspace-name-pipelines-pipeline-name-chat-stream-post.StatusCodes.json")}
>
  
</StatusCodes>