Stream Pipeline Responses with the REST API

Get pipeline output as it is generated with the Search Stream and Chat Stream endpoints, and handle the server-sent events.

---

Prerequisites

• You need a Haystack Platform API key. For details, see Generate API Keys.
• You need a deployed, active pipeline.
• To stream output from more than one component, configure streaming_components in your pipeline YAML. For details, see Enable Streaming.

Send a Streaming Request

You can use streaming with the stream API endpoints: Chat Stream and Search Stream. This is an example request to the Search Stream endpoint.

cURL

  curl --request POST \
      --url https://api.cloud.deepset.ai/api/v1/workspaces/<YOUR_WORKSPACE>/pipelines/<YOUR_PIPELINE>/search-stream \
      --header 'accept: application/json' \
      --header 'authorization: Bearer <YOUR_API_KEY>' \
      --header 'content-type: application/json' \
      --data '
  {
  "debug": false,
  "include_result": true,
  "view_prompts": false,
  "query": "who started all-girl bands?"
  }
  '

Python

  import requests

  url = "https://api.cloud.deepset.ai/api/v1/workspaces/<YOUR_WORKSPACE>/pipelines/<YOUR_PIPELINE>/search-stream"

  payload = {
      "debug": False,
      "include_result": True,
      "view_prompts": False,
      "query": "who started all-girl bands?"
  }
  headers = {
      "accept": "application/json",
      "content-type": "application/json",
      "authorization": "Bearer <YOUR_API_KEY>"
  }

  response = requests.post(url, json=payload, headers=headers)

  print(response.text)

Replace these placeholders:

• <YOUR_WORKSPACE>: With the name of the workspace containing your pipeline.
• <YOUR_PIPELINE>: With the name of the pipeline to use for search.
• <YOUR_API_KEY>: With your Haystack Platform API key.

Default Streaming Behavior in the API

By default, the streaming endpoints only stream output from the last streaming-capable component in your pipeline. This matches the default pipeline behavior described in the Default Behavior section of the Enable Streaming guide.

To stream output from more than one component or from a specific component, configure the streaming_components field in your pipeline YAML before using the API. See Enable Streaming for Specific Components for details.

Stream Request Parameters

These request body fields control which SSE event types you receive from search-stream and chat-stream:

Parameter	Default	Effect
include_result	false	When true, emits a result event with the full pipeline output at the end of the stream
include_tool_calls	rendered	Controls how agent tool calls appear in the stream (see below)
include_tool_call_results	false	When true, emits tool_call_result events after each tool completes
include_reasoning	false	When true, emits reasoning events during model reasoning steps
allow_sync_streaming_callbacks	true	When true, enables Hayhooks hybrid mode for pipelines that use legacy sync-only streaming_callback components
chat_history_granularity	QUERY_ANSWER	Controls how much chat history is replayed to agent pipelines on each turn (see below)

include_tool_calls

The default is rendered. This setting only affects agent pipelines that make tool calls:

Value	Behavior
rendered (default)	Tool calls are rendered as inline markdown inside delta events. No tool_call_delta events are emitted.
true	Tool calls emit structured tool_call_delta events as they progress
false	Tool calls are omitted from the stream entirely

The rendered default matches Playground behavior and is the simplest option for chat-style UIs. Set include_tool_calls to true when you need programmatic access to tool names, arguments, and call IDs with tool_call_delta events.

chat_history_granularity

This setting controls how much of the conversation history is sent back to the pipeline on each turn. It is most relevant for agent pipelines, where intermediate tool calls, tool results, and reasoning steps are part of the conversation.

Value	Behavior
QUERY_ANSWER (default)	Only the user query and the final assistant answer from each previous turn are included in the history.
ALL_MESSAGES	The full message trace from each previous turn is included — tool calls, tool results, reasoning steps, and the final answer.

Use ALL_MESSAGES when your agent pipeline relies on tool call context across multiple turns. For example, if a user follows up on a previous tool-assisted response, ALL_MESSAGES makes sure the agent receives the complete prior context and does not lose track of earlier tool use.

Token Usage

Using ALL_MESSAGES increases the amount of context sent to the model on each turn. This can increase token usage and costs compared to the default QUERY_ANSWER mode.

You can also set this per pipeline in the pipeline YAML under the history key:

history:
  granularity: all_messages

When both the API request and the pipeline YAML specify a value, the API request value takes precedence. If neither is set, the default is QUERY_ANSWER.

Streaming Event Types

The streaming endpoints use Server-Sent Events (SSE) and return a sequence of events. Each event has a type field that identifies what it contains. Most events also include a query_id. The ping event is the only exception. The event types are mutually exclusive, each event carries exactly one of the following:

Event type	When it's sent	What it contains
delta	During generation	delta object with text and meta; optional start (first chunk from a component), finish_reason (when generation ends), and index (component index when multiple components stream)
result	End of stream (when include_result=true)	The full pipeline result
error	When the pipeline fails	An error message. Terminal — no further events follow it.
tool_call_delta	During agent tool use (only when include_tool_calls=true)	Structured information about a tool call in progress. Not emitted when the default "rendered" mode is used
tool_call_result	After a tool call completes (when include_tool_call_results=true)	The result returned by a tool
reasoning	During agent reasoning (when include_reasoning=true)	A reasoning step from the model
ping	Periodically, and during infrastructure warmup	Keep-alive signal. Contains only type — no query_id. May include an optional phase field set to cold_start while temporary infrastructure and pipeline preparation is still running

The delta Event

delta events carry incremental text as the pipeline generates a response. The delta object holds the new text and metadata (including which component produced the chunk). Three optional top-level fields help you format output across multiple streaming components:

Field	Type	When present	Meaning
start	boolean	First chunk from a component	true marks the beginning of output from that component
finish_reason	string	Last chunk from a component	Why generation stopped: stop, length, tool_calls, content_filter, or tool_call_results
index	integer	Multiple streaming components	Identifies which content part is being updated

{
  "type": "delta",
  "query_id": "290a1f96-57d6-4843-8ed7-2a224142398b",
  "delta": {
    "text": "Hello",
    "meta": {
      "deepset_cloud": {
        "component": "llm_1"
      }
    }
  },
  "start": true,
  "finish_reason": "stop",
  "index": 0
}

start, finish_reason, and index are omitted when they do not apply to a given chunk.

The error event

Every failed stream ends with an error event. The event includes a human-readable error message. The error frame is terminal — no further events follow it. This also holds true after you receive a ping event: if a failure happens once streaming has started, it arrives as a single terminal error event and is not followed by a result event.

{
  "type": "error",
  "query_id": "290a1f96-57d6-4843-8ed7-2a224142398b",
  "error": "Invalid pipeline configuration."
}

The ping event

The server sends ping events periodically to keep the connection alive during long-running streams. Unlike every other event type, ping carries no query_id — only the type field:

{
  "type": "ping"
}

A ping event can also include an optional phase field. When phase is set to cold_start, it tells you that temporary infrastructure and pipeline preparation is still running behind the scenes, for example right after a new replica has scaled up. It is not answer text, and you can treat it the same way as a plain ping:

{
  "type": "ping",
  "phase": "cold_start"
}

The phase value is extensible, so new values may be added in the future. You can safely ignore ping events, and any phase value you do not recognize, in your client logic.

Here is an example showing how to handle all event types. The example sets include_tool_calls to true so that structured tool_call_delta events are emitted — omit this field to use the default "rendered" mode, where tool calls appear as markdown inside delta events instead:

import httpx
import json
from httpx_sse import EventSource
import asyncio

TOKEN = "<YOUR_API_KEY>"
PIPELINE_URL = "https://api.cloud.deepset.ai/api/v1/workspaces/<YOUR_WORKSPACE>/pipelines/<YOUR_PIPELINE>"


async def main():
    query = {
        "query": "How does streaming work with deepset?",
        "include_result": True,
        # Default is "rendered" (tool calls as markdown in delta events).
        # Set to true to receive structured tool_call_delta events instead:
        "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", "/search-stream", json=query) as response:
            event_source = EventSource(response)
            async for sse in event_source.aiter_sse():
                event_data = json.loads(sse.data)
                match event_data["type"]:
                    case "delta":
                        print(event_data["delta"]["text"], flush=True, end="")
                    case "result":
                        print("\n\nFull result:")
                        print(json.dumps(event_data["result"]))
                    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())

Related Information

• Enable Streaming
• Run a Search
• Chat Stream API reference
• Search Stream API reference