[Feature Request] Add cancellation API for streaming callbacks · Issue #4004 · plotly/dash · GitHub
Skip to content

[Feature Request] Add cancellation API for streaming callbacks #4004

Description

@c0debeez

Thanks so much for your interest in Dash!

Before posting an issue here, please check the Dash community forum to see if the topic has already been discussed. The community forum is also great for implementation questions. When in doubt, please feel free to just post the issue here :)

Is your feature request related to a problem? Please describe.

Dash 4.5 streaming callbacks allow an async def generator callback to stream yielded values to the browser, including incremental Patch updates. However, an application cannot currently provide a reliable stop button for an ordinary streaming callback through the public callback API.

running can keep start/stop controls in the appropriate loading and disabled states, but it does not cancel the active generator. Triggering the callback again also does not reliably stop the previous invocation. The existing cancel=[Input(...)] callback argument is limited to background callbacks.

The transport already has cancellation primitives, but they are not exposed to applications:

  • With multiplexed streaming, the renderer tracks a per-stream request ID and can send streamCancel; this is already used when a tab is closed.
  • Without shared storage, each stream falls back to its own NDJSON HTTP response. Aborting that request can close the response iterator and cancel the async generator, but application code has no public handle for the request or its AbortController.

As a result, applications must build a separate cooperative cancellation protocol, often involving ctx.shared_storage, unique stream IDs, polling a cancellation flag before every yield, and manual cleanup. This is much more application code than should be necessary for a fundamental streaming operation.

Describe the solution you'd like

Please support cancel=[Input(...)] for ordinary streaming callbacks, with semantics similar to background callback cancellation:

from dash import Input, Output, Patch, callback

@callback(
    Output("out", "children"),
    Input("start", "n_clicks"),
    cancel=[Input("stop", "n_clicks")],
    running=[
        (Output("start", "loading"), True, False),
        (Output("start", "disabled"), True, False),
        (Output("stop", "disabled"), False, True),
    ],
    prevent_initial_call=True,
)
async def stream(_):
    async for token in llm.stream(prompt):
        patch = Patch()
        patch += token
        yield patch

Ideally, Dash would implement cancellation according to the active transport:

  • Standalone NDJSON: abort the callback fetch request with its AbortController.
  • Multiplexed SharedWorker transport: send streamCancel for the active request ID without closing the shared downlink.
  • Server: cancel/close the active async generator, run its finally cleanup, restore all running outputs, and treat cancellation as expected control flow rather than a callback error.

This should work without requiring application-level polling or direct use of shared storage. If multiple invocations are permitted concurrently, the API should define whether the cancel input cancels all active invocations for that callback or only the invocation associated with the current output grouping.

Describe alternatives you've considered

  • Store cancellation flags in ctx.shared_storage and check them before every yielded item. This works across workers but requires stream IDs, polling, cleanup, and storage configuration in application code.
  • Use an in-process dict[str, asyncio.Event]. This works only in a single-process deployment and fails when the start and stop requests reach different workers.
  • Build a custom clientside component that retains the request's AbortController. This depends on renderer internals and duplicates transport logic.
  • Use a background callback only to obtain cancel=[Input(...)]. Background callbacks do not provide the same incremental streaming behavior.
  • Replace the native callback with EventSource or Socket.IO and implement a separate cancellation endpoint. This gives up the simplicity of native Dash streaming callbacks.

Additional context

Streaming callbacks were introduced in #3931. The existing SharedWorker/page-close cancellation path demonstrates that Dash already has most of the transport and server-side cancellation machinery. Exposing it through the callback API would make native streaming callbacks suitable for common LLM interfaces with explicit Start generating and Stop generating controls, while preserving the single shared browser connection used by multiplexed streams.

Activity

  1. camdecoster commented on Sep 22, 2026

    @camdecoster
    Contributor

    Thanks for the request! This should be possible. It would probably be best for our team to add a feature like this, but we're open to ideas from the community.

  2. T4rk1n commented on Sep 24, 2026

    @T4rk1n
    Contributor

    Until cancel= works for streaming callbacks, here's a workaround using shared storage pub/sub. The streaming callback subscribes to a stop topic and races it against the token source. The Stop button publishes to that topic. There's no polling and no flags, and it works across workers.

    import asyncio
    import uuid
    
    from dash import Dash, Input, Output, Patch, State, ctx, dcc, html
    
    app = Dash(__name__)
    app.layout = lambda: html.Div([
        dcc.Store(id="session-id", data=uuid.uuid4().hex),
        html.Button("Start generating", id="start"),
        html.Button("Stop generating", id="stop", disabled=True),
        html.Div(id="out"),
    ])
    
    
    async def until_stopped(source, stop_sub):
        """Yield from source until a message arrives on stop_sub."""
        stop_task = asyncio.ensure_future(stop_sub.__aiter__().__anext__())
        item_task = None
        try:
            while True:
                item_task = asyncio.ensure_future(source.__anext__())
                await asyncio.wait({item_task, stop_task}, return_when=asyncio.FIRST_COMPLETED)
                if not item_task.done():
                    return
                try:
                    item = item_task.result()
                except StopAsyncIteration:
                    return
                yield item
        finally:
            # source can't be closed while a __anext__ on it is in flight
            tasks = [t for t in (item_task, stop_task) if t is not None]
            for t in tasks:
                t.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            stop_sub.close()
            await source.aclose()
    
    
    def stop_topic(session_id, run):
        return f"stop:{session_id}:{run}"
    
    
    @app.callback(
        Output("out", "children"),
        Input("start", "n_clicks"),
        State("session-id", "data"),
        running=[
            (Output("start", "disabled"), True, False),
            (Output("stop", "disabled"), False, True),
        ],
        prevent_initial_call=True,
    )
    async def generate(run, session_id):
        # Stop is enabled before the request is sent, so replay from the start
        # of this run's topic to catch an early click.
        stop_sub = ctx.shared_storage.subscribe(stop_topic(session_id, run), replay_from=0)
        yield ""
        async for token in until_stopped(llm.stream(prompt), stop_sub):
            patch = Patch()
            patch += token
            yield patch
    
    
    @app.callback(
        Input("stop", "n_clicks"),
        State("session-id", "data"),
        State("start", "n_clicks"),
        prevent_initial_call=True,
    )
    def stop(_, session_id, run):
        ctx.shared_storage.publish(stop_topic(session_id, run), True)

    On Stop, the source generator is closed, so its finally cleanup runs, and the running outputs restore the buttons.

    Caveats:

  3. added theissue type on Sep 24, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions