Usage

Streaming

A reply that is sent while it is still being written. Posts the first chunk, then edits that same message as more text arrives. Falls back to a single post on close when the runtime or channel cannot edit — handler code never branches on it.

@cx.on_message({"channel": "telegram", "command": "story"})
def on_story(thread, msg, ctx):
    with thread.stream(min_chars=1, throttle=0.25) as out:
        for chunk in llm:
            out.append(chunk)
    # __exit__ always close()s so the final flush happens

const out = thread.stream({ minChars: 1, throttle: 0.25 })
for await (const chunk of llm) {
  await out.append(chunk)
}
await out.close()

Options

	Default
`min_chars`	24	Do not flush until this many new characters accumulated.
`throttle`	0.5s	At most one edit per interval. Platforms rate-limit per-token edits. `close()` always sends the complete text.

live vs buffered

out.live is true when chunks go out as they arrive. If the sink cannot stream (no message id back, or the channel cannot edit), further appends buffer and close() does one post.

Overlap while streaming

thread.stream is how you write the reply. Overlap is whether a second event may start a handler while that stream is still open. See Overlap.

# Python: overlap "stream" always executes (same FSM as parallel).
@cx.on_action({"channel": "telegram", "data": "story", "overlap": "stream"})
def on_story(thread, action, ctx):
    with thread.stream(min_chars=1, throttle=0.25) as out:
        for chunk in STORY:
            out.append(chunk)

# Voice / barge-in: drop the new turn; do not queue behind speech.
@cx.on_message({"channel": "voice", "overlap": "drop"})
def on_speech(thread, msg, ctx):
    thread.post(msg.text)

// TypeScript has no "stream" policy yet — use parallel (always execute).
cx.onAction({ channel: "telegram", data: "story", overlap: "parallel" }, async (thread) => {
  const out = thread.stream({ minChars: 1, throttle: 0.25 })
  for (const chunk of STORY) await out.append(chunk)
  await out.close()
})

// Voice / barge-in: drop the new turn; do not queue behind speech.
cx.onMessage({ channel: "voice", overlap: "drop" }, async (thread, msg) => {
  await thread.post(msg.text)
})