Streaming

Streaming lets you post a placeholder message and update it incrementally as content arrives, similar to how ChatGPT shows tokens appearing. This is useful for AI-powered bots that generate responses token by token.

Basic UsageCopy Anchor Link

bot.on_new_mention do |thread, message|
  thread.post_stream(placeholder: "Thinking...") do |stream|
    # Simulate token-by-token generation
    words = ["Hello", " ", "world", "!", " ", "How", " ", "are", " ", "you", "?"]
    words.each do |word|
      stream << word
      sleep(0.1)
    end
  end
end

post_stream returns the posted ChatSDK::Message. It also accepts an enumerable of chunks:

message = thread.post_stream(ai_client.stream(prompt))

How It WorksCopy Anchor Link

The block receives a writer that accepts chunks with <<. The SDK normalizes text and structured chunks, then uses an adapter's optional native streaming hook when available. If the hook returns nil, the core posts and edits a message at the configured streaming_update_interval (default: 0.5 seconds).

Structured chunks are represented by ChatSDK::Streaming::MarkdownTextChunk, TaskUpdateChunk, and PlanUpdateChunk. The core edit-based fallback includes markdown text and ignores task and plan chunks.

chunks = [
  ChatSDK::Streaming::MarkdownTextChunk.new(text: "Working..."),
  ChatSDK::Streaming::TaskUpdateChunk.new(
    id: "lookup",
    title: "Look up records",
    status: "in_progress"
  ),
  ChatSDK::Streaming::PlanUpdateChunk.new(title: "Review results")
]
thread.post_stream(chunks)

ChatSDK::StreamingPlan carries task grouping, end blocks, update interval, and AI-generated metadata:

plan = ChatSDK::StreamingPlan.new(
  chunks,
  group_tasks: :plan,
  end_with: [{type: "context", elements: []}],
  ai_generated: true
)
thread.post(plan)

group_tasks accepts :plan or :timeline. Task statuses are pending, in_progress, complete, and error.

ConfigurationCopy Anchor Link

The throttle interval is controlled by streaming_update_interval (default: 0.5 seconds):

bot = ChatSDK::Chat.new(
  user_name: "my-bot",
  adapters: { slack: slack },
  state: state,
  streaming_update_interval: 1.0  # Flush at most once per second
)

Without a PlaceholderCopy Anchor Link

If you omit the placeholder:, the first flush creates the initial message:

thread.post_stream do |stream|
  stream << "Starting..."
  # The message is created on the first flush
  stream << " done!"
end

Adapter CompatibilityCopy Anchor Link

Adapters may implement the optional Adapter::Base#stream hook for native streaming. When the hook returns nil without consuming the source, ChatSDK uses its post-and-edit fallback. If the adapter cannot edit messages, it posts the final accumulated text once.

Slack Native StreamingCopy Anchor Link

Slack native streaming requires a thread ID and either a direct-message channel or both recipient user and team IDs. The adapter derives recipient IDs from the current inbound message when available. Otherwise, the post-and-edit fallback is used.

slack = ChatSDK::Slack::Adapter.new(
  bot_token: ENV["SLACK_BOT_TOKEN"],
  signing_secret: ENV["SLACK_SIGNING_SECRET"],
  native_streaming: true,
  stream_segment_max_age: 240
)

native_streaming: false disables the native API. stream_segment_max_age controls when long streams rotate into a new Slack message; it must be positive, and Float::INFINITY disables rotation. If Slack rejects the native API before any message is rendered, ChatSDK falls back to post-and-edit. Errors after native output has rendered are raised.

Task and plan chunks use Slack's Assistants streaming APIs and require the assistant:write scope. Plain markdown streaming does not require this scope.

Integrating with AI APIsCopy Anchor Link

Here is a realistic example using a streaming AI response:

bot.on_new_mention do |thread, message|
  thread.post_stream(placeholder: "Generating response...") do |stream|
    client = OpenAI::Client.new
    client.chat(
      parameters: {
        model: "gpt-4",
        messages: [{ role: "user", content: message.text }],
        stream: proc do |chunk|
          content = chunk.dig("choices", 0, "delta", "content")
          stream << content if content
        end
      }
    )
  end
end

Testing StreamsCopy Anchor Link

Use FakeAdapter to verify streaming behavior:

bot = ChatSDK::Testing.build_bot
adapter = bot.config.adapters[:test]

bot.on_new_mention do |thread, message|
  thread.post_stream(placeholder: "...") do |stream|
    stream << "hello "
    stream << "world"
  end
end

adapter.simulate_mention(bot, text: "test")

# The placeholder is posted, then edited with the final content
assert adapter.posted_messages.size >= 1
assert adapter.edited_messages.size >= 1