Build your first agent in just a few minutes with [Rasa Copilot](https://hello.rasa.com/?utm_source=docs&utm_medium=referral&utm_campaign=docs_cta). ## CollectingDispatcher A dispatcher is an instance of the `CollectingDispatcher` class used to generate responses to send back to the user. ### CollectingDispatcher `CollectingDispatcher` provides two complementary APIs: - **Unary API** — `utter_message` accumulates complete messages in the `messages` attribute. The action server returns them in the final response payload. - **Streaming API** — `stream_start`, `stream_chunk`, and `stream_end` let an action emit text or rich content incrementally. Whether chunks are delivered in real time is decided by the transport layer (see [Streaming responses](/content/docs/reference/integrations/action-server/sdk-dispatcher/#streaming-responses/index.html)), not by the action author. In both cases, the Rasa server adds `BotUttered` events to the tracker for responses the user received. Responses added using the dispatcher should therefore not be returned explicitly as [events](/content/docs/reference/integrations/action-server/events/index.html). For example, the following custom action returns no events explicitly but will return the response, "Hi, User!" to the user: ```python class ActionGreetUser(Action): def name(self) -> Text: return "action_greet_user" async def run( self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any], ) -> List[EventType]: dispatcher.utter_message(text = "Hi, User!") return [] ``` ### CollectingDispatcher.utter_message The `utter_message` method can be used to return any type of response to the user. #### **Parameters** The `utter_message` method takes the following optional arguments. Passing no arguments will result in an empty message being returned to the user. Passing multiple arguments will result in a rich response (e.g. text and buttons) being returned to the user. - `text`: The text to return to the user. ```python dispatcher.utter_message(text = "Hey there") ``` - `image`: An image URL or file path that will be used to display an image to the user. ```python dispatcher.utter_message(image = "https://i.imgur.com/nGF1K8f.jpg") ``` - `json_message`: A custom json payload as a dictionary. It can be used to send [channel specific responses](/content/docs/reference/primitives/responses/index.html). ```python date_picker = { "blocks":[\ {\ "type": "section",\ "text":{\ "text": "Make a bet on when the world will end:",\ "type": "mrkdwn"\ },\ "accessory":\ {\ "type": "datepicker",\ "initial_date": "2019-05-21",\ "placeholder":\ {\ "type": "plain_text",\ "text": "Select a date"\ }\ }\ }\ ] } dispatcher.utter_message(json_message = date_picker) ``` - `response`: The name of a response to return to the user. This response should be specified in your assistants [domain](/content/docs/reference/config/domain/index.html). ```python dispatcher.utter_message(response = "utter_greet") ``` - `attachment`: A URL or file path of an attachment to return to the user. ```python dispatcher.utter_message(attachment = "") ``` - `buttons`: A list of buttons to return to the user. Each button is a dictionary and should have a `title` and a `payload` key. A button can include other keys, but these will only be used if a specific channel looks for them. The button's `payload` will be sent as a user message if the user clicks the button. ```text dispatcher.utter_message(buttons = [\ {"payload": "/affirm", "title": "Yes"},\ {"payload": "/deny", "title": "No"},\ ]) ``` - `elements`: These are specific to using Facebook as a messaging channel. For details of expected format see [Facebook's documentation](https://developers.facebook.com/docs/messenger-platform/send-messages/template/generic/) - `**kwargs`: arbitrary keyword arguments, which can be used to specify values for [variable interpolation in response variations](/content/docs/reference/primitives/responses/index.html). For example, given the following response: ```yaml responses: utter_greet_name: - text: Hi {name}! ``` You could specify the name with: ```python dispatcher.utter_message(response = "utter_greet_name", name = "Aimee") ``` #### **Return type** `None` ## Streaming responses Custom actions can stream responses token-by-token or as a sequence of rich-content fragments. Use the streaming API when you want low-latency delivery on channels that support it — for example, voice assistants synthesizing TTS from partial text, or messaging channels that render incremental assistant output. ### Supported transports Real-time streaming is available only when **both** of the following are true: 1. The action runs on a **streaming-capable executor** — [direct custom action execution](/content/docs/reference/primitives/custom-actions/#running-custom-actions-directly-by-the-assistant/index.html) (`actions_module`) or a [gRPC action server](/content/docs/reference/integrations/action-server/actions/#grpc-protocol/index.html). 2. The output channel supports streaming — see [Built-in output channels with streaming support](/content/docs/reference/channels/custom-connectors/#built-in-output-channels-with-streaming-support/index.html). The **HTTP(S) action server protocol does not support streaming**. When Rasa calls an HTTP action server, it sends a single HTTP POST to the `/webhook` endpoint and waits for one JSON response — even if the output channel could otherwise stream. Your action code can still use `stream_start`, `stream_chunk`, and `stream_end`; the SDK accumulates chunks internally and replays them as individual `utter_message` calls when `stream_end()` runs, so the user receives the full response once the action completes — not incrementally. No conditional logic is required in the action. ### Writing a streaming action Call `stream_start()` before emitting chunks, `stream_chunk()` for each fragment, and `stream_end()` when the response is complete: ```python class ActionStreamReply(Action): def name(self) -> Text: return "action_stream_reply" async def run( self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any], ) -> List[EventType]: await dispatcher.stream_start() async for token in fetch_tokens_from_llm(tracker.latest_message.text): await dispatcher.stream_chunk(text=token) await dispatcher.stream_end() return [] ``` You can also stream rich content mid-response — for example, append buttons after the main text: ```python await dispatcher.stream_start() async for token in fetch_tokens_from_llm(prompt): await dispatcher.stream_chunk(text=token) await dispatcher.stream_chunk(buttons=[\ {"title": "Yes", "payload": "/affirm"},\ {"title": "No", "payload": "/deny"},\ ]) await dispatcher.stream_end() ``` If you call `stream_chunk()` without an explicit `stream_start()`, the dispatcher opens a stream implicitly. If an action opens a stream but forgets to call `stream_end()`, the executor closes the stream automatically and logs a warning. ### CollectingDispatcher.stream_start ```python await dispatcher.stream_start() ``` Begins a streaming response and resets the internal chunk accumulator. When a streaming transport is active, emits a `stream_start` event to the caller. Calling `stream_start()` again before `stream_end()` resets the accumulator and starts a new stream sequence. #### **Return type** `None` ### CollectingDispatcher.stream_chunk ```python await dispatcher.stream_chunk( text=None, image=None, json_message=None, attachment=None, buttons=None, elements=None, **kwargs, ) ``` Emits a response fragment. Accepts the same rich-content fields as `utter_message`, **except**`template` and `response` — those refer to pre-defined domain responses and are not valid in a streaming context. Passing either raises `ValueError`. #### **Parameters** - `text`: A plain-text fragment (for example, a token from an LLM). - `image`: URL of an image to include in this chunk. - `json_message`: Arbitrary custom JSON payload for this chunk. - `attachment`: URL of an attachment to include in this chunk. - `buttons`: List of button dicts to include in this chunk. - `elements`: List of carousel/card element dicts for this chunk. - `**kwargs`: Extra fields merged into the chunk payload. Unlike the named parameters above, these are merged unconditionally (no `None`/falsy guard). On **gRPC streaming**, only fields defined in the `Chunk` protobuf message are serialised and sent to Rasa (`text`, `image`, `custom`, `attachment`, `buttons`, `elements`). Any other keys in `**kwargs` are **silently omitted** from the stream. This is intentional: the protobuf schema must describe fields precisely, and without a concrete use case for arbitrary extras the SDK exposes only known parameters on `CollectingDispatcher`. On **non-streaming transports** (including HTTP action servers), each accumulated chunk is replayed via `utter_message()` at `stream_end()`, and extra keys survive that replay. Callers should pass only keys the target transport understands. #### **Return type** `None` ### CollectingDispatcher.stream_end ```python await dispatcher.stream_end() ``` Ends the streaming response. On streaming transports, emits a `stream_end` event. Chunks were already delivered in-band, so the dispatcher does **not** replay them via `utter_message()`. On non-streaming transports, replays every accumulated chunk as an individual `utter_message()` call so the tracker receives a record of the full response. #### **Return type** `None` ### CollectingDispatcher.is_streaming_active Read-only property. Returns `True` after `stream_start()` has been called and before `stream_end()` completes. The executor uses this to detect actions that opened a stream but never closed it. ### Barge-in and stream cancellation When a user interrupts the assistant while a streaming custom action is in progress (voice barge-in), the transport layer signals the dispatcher to stop producing output. - **gRPC action server** — Rasa sends an `AckStreamChunks` RPC with the active `response_id`. The SDK calls `cancel_stream()` on the matching dispatcher. - **Direct custom action executor** — Rasa sets the same cancellation flag on the in-process dispatcher. `cancel_stream()` is called by the transport layer. Action authors do not call it directly. ### CollectingDispatcher.cancel_stream Sets `is_streaming_cancelled` to `True`. All subsequent `stream_chunk()` calls are silently dropped so the action stops producing output without being forcefully interrupted. The action continues running until its `run()` method returns, so events such as `SlotSet` can still be applied. ### CollectingDispatcher.is_streaming_cancelled Read-only property. Returns `True` after `cancel_stream()` has been called. You can inspect this after `stream_end()` to suppress side effects that are irrelevant when the user already interrupted: ```python await dispatcher.stream_end() if dispatcher.is_streaming_cancelled: return [] return [SlotSet("result", value)] ``` Checking the flag is optional. If you do not inspect it, all events returned by the action are preserved — which is the right default for most actions.