How to Implement a Cancellable/Interruptible Streaming Event Handler?

[jean-fox]

Hi!

First, thanks for creating such a great framework. I'm trying to implement a common UI pattern: a long-running streaming task (e.g., from an LLM API) that the user can cancel on-demand by clicking a "Cancel" button.

Through experimentation, I've found that Mesop's event model appears to be strictly serial for a given user session. While an async event handler is running its task, any other events (like a click on a "Cancel" or "Test" button) are queued and only processed after the initial streaming handler has fully completed. This makes it impossible to interrupt the running stream via another standard Mesop event.

The attached codes, simple_test.py, is a minimal, self-contained, reproducible example that demonstrates this exact behavior. If you run this example and click "Start Stream", you will notice that the "Test Click" button is unresponsive until the stream finishes, at which point all the queued click events are processed at once.

My question is: What is the idiomatic or recommended architecture in Mesop for implementing a long-running, streaming task that can be cancelled on demand by another user action?

Given the serial event processing model, it seems the solution must involve a pattern that runs the task outside of the primary event handler's lifecycle, but I'm unsure of the best-practice way to achieve this within the framework. Any guidance or a simple example demonstrating the recommended pattern would be incredibly helpful.

Thank you for your time and for all your work on Mesop!

Attachment: simple_test.py

import mesop as me
import asyncio

@me.stateclass
class State:
    text: str = ""
    is_streaming: bool = False

async def stream_generator():
    """A simple async generator that slowly yields numbers."""
    for i in range(1, 16):
        yield f"Chunk {i}... "
        # Using a 1-second sleep to make it obvious that the UI
        # is unresponsive to other events during this await.
        await asyncio.sleep(1)

async def on_start_stream(e: me.ClickEvent):
    """
    This is the long-running streaming event handler.
    """
    state = me.state(State)
    state.is_streaming = True
    state.text = ""
    yield # Immediately update the UI to disable the button.

    # Iterate through the async generator.
    async for chunk in stream_generator():
        state.text += chunk
        # Yield after each chunk to update the UI.
        yield

    # Update state after the stream is finished.
    state.is_streaming = False
    yield

def on_test_click(e: me.ClickEvent):
    """
    A simple sync handler to test if the UI is blocked.
    """
    print("--- <SUCCESS> Test Button was clicked! UI is NOT blocked. ---")

@me.page(path="/",
         security_policy=me.SecurityPolicy(dangerously_disable_trusted_types=True))
def page():
    """Page definition."""
    state = me.state(State)
    me.button("Start Stream", on_click=on_start_stream, disabled=state.is_streaming)
    me.button("Test Click", on_click=on_test_click)
    me.divider()
    me.text(state.text)

[jean-fox]

I have found the solution: run mesop in WebSockets mode by setting MESOP_WEBSOCKETS_ENABLED=true

Thx!

[richard-to]

Awesome. Glad you found a solution.