I tried the new Slack API methods [Streaming Edition]

# Slack Text Streaming API Guide ## Overview The Slack Text Streaming API provides three methods for sequentially updating blocks using the `chunks` parameter: `chat.startStream`, `chat.appendStream`, and `chat.stopStream`. --- ## API Methods ### 1. `chat.startStream` Initializes a new streaming session and returns a `stream_ts` (stream timestamp) used in subsequent calls. **Parameters:** | Parameter | Type | Required | Description | |-----------|------|----------|-------------| | `channel` | string | Yes | Target channel ID | | `assistant_color` | string | No | Color of the assistant indicator | | `title` | string | No | Title displayed above the stream | **Example Request:** ```json POST https://slack.com/api/chat.startStream { "channel": "C0123456789", "title": "AI Response" } ``` **Example Response:** ```json { "ok": true, "stream_ts": "1710000000.000100", "channel": "C0123456789" } ``` --- ### 2. `chat.appendStream` Appends content chunks to an existing stream. This is the core method for sending text incrementally. **Parameters:** | Parameter | Type | Required | Description | |-----------|------|----------|-------------| | `channel` | string | Yes | Target channel ID | | `stream_ts` | string | Yes | Stream timestamp from `startStream` | | `chunks` | array | Yes | Array of block objects to append | | `cursor_visibility` | string | No | `visible` or `hidden` | | `new_message_ts` | string | No | Thread timestamp if replying | **`chunks` Parameter Structure:** ```json { "chunks": [ { "type": "rich_text", "elements": [ { "type": "rich_text_section", "elements": [ { "type": "text", "text": "Hello, this is streamed text!" } ] } ] } ] } ``` **Example Request:** ```json POST https://slack.com/api/chat.appendStream { "channel": "C0123456789", "stream_ts": "1710000000.000100", "cursor_visibility": "visible", "chunks": [ { "type": "rich_text", "elements": [ { "type": "rich_text_section", "elements": [ { "type": "text", "text": "Generating response..." } ] } ] } ] } ``` **Example Response:** ```json { "ok": true, "stream_ts": "1710000000.000100" } ``` --- ### 3. `chat.stopStream` Finalizes the stream and marks it as complete. The cursor is hidden after this call. **Parameters:** | Parameter | Type | Required | Description | |-----------|------|----------|-------------| | `channel` | string | Yes | Target channel ID | | `stream_ts` | string | Yes | Stream timestamp to finalize | | `cursor_visibility` | string | No | Typically set to `hidden` | **Example Request:** ```json POST https://slack.com/api/chat.stopStream { "channel": "C0123456789", "stream_ts": "1710000000.000100", "cursor_visibility": "hidden" } ``` **Example Response:** ```json { "ok": true, "stream_ts": "1710000000.000100" } ``` --- ## End-to-End Implementation Example ### Python Implementation ```python import time import requests SLACK_TOKEN = "xoxb-your-bot-token" CHANNEL_ID = "C0123456789" headers = { "Authorization": f"Bearer {SLACK_TOKEN}", "Content-Type": "application/json" } def start_stream(channel: str, title: str = "") -> str: """Start a new stream and return stream_ts.""" payload = { "channel": channel, "title": title } response = requests.post( "https://slack.com/api/chat.startStream", headers=headers, json=payload ) data = response.json() if not data.get("ok"): raise Exception(f"Failed to start stream: {data.get('error')}") return data["stream_ts"] def append_stream(channel: str, stream_ts: str, text: str) -> None: """Append a text chunk to the stream.""" payload = { "channel": channel, "stream_ts": stream_ts, "cursor_visibility": "visible", "chunks": [ { "type": "rich_text", "elements": [ { "type": "rich_text_section", "elements": [ { "type": "text", "text": text } ] } ] } ] } response = requests.post( "https://slack.com/api/chat.appendStream", headers=headers, json=payload ) data = response.json() if not data.get("ok"): raise Exception(f"Failed to append stream: {data.get('error')}") def stop_stream(channel: str, stream_ts: str) -> None: """Finalize and close the stream.""" payload = { "channel": channel, "stream_ts": stream_ts, "cursor_visibility": "hidden" } response = requests.post( "https://slack.com/api/chat.stopStream", headers=headers, json=payload ) data = response.json() if not data.get("ok"): raise Exception(f"Failed to stop stream: {data.get('error')}") def stream_text_to_slack(channel: str, full_text: str, chunk_size: int = 10): """Stream text to Slack in chunks.""" # Step 1: Start the stream stream_ts = start_stream(channel, title="AI Response") print(f"Stream started: {stream_ts}") try: # Step 2: Append text in chunks words = full_text.split() buffer = "" for i, word in enumerate(words): buffer += word + " " # Send every `chunk_size` words if (i + 1) % chunk_size == 0: append_stream(channel, stream_ts, buffer) print(f"Appended: {buffer.strip()}") buffer = "" time.sleep(0.3) # Simulate streaming delay # Send any remaining text if buffer.strip(): append_stream(channel, stream_ts, buffer) print(f"Appended final chunk: {buffer.strip()}") finally: # Step 3: Stop the stream stop_stream(channel, stream_ts) print("Stream stopped.") # --- Run the example --- sample_text = ( "This is a demonstration of the Slack Text Streaming API. " "Text is sent incrementally in chunks, simulating a real-time " "AI response being generated token by token." ) stream_text_to_slack(CHANNEL_ID, sample_text, chunk_size=5) ``` --- ## Validation Example ### Testing Each API Method ```python import unittest from unittest.mock import patch, MagicMock class TestSlackStreamingAPI(unittest.TestCase): @patch("requests.post") def test_start_stream_success(self, mock_post): """Verify startStream returns a valid stream_ts.""" mock_response = MagicMock() mock_response.json.return_value = { "ok": True, "stream_ts": "1710000000.000100", "channel": "C0123456789" } mock_post.return_value = mock_response stream_ts = start_stream("C0123456789", title="Test Stream") self.assertEqual(stream_ts, "1710000000.000100") mock_post.assert_called_once() call_args = mock_post.call_args self.assertIn("chat.startStream", call_args[0][0]) print("✅ test_start_stream_success passed") @patch("requests.post") def test_append_stream_success(self, mock_post): """Verify appendStream sends chunks correctly.""" mock_response = MagicMock() mock_response.json.return_value = { "ok": True, "stream_ts": "1710000000.000100" } mock_post.return_value = mock_response append_stream("C0123456789", "1710000000.000100", "Hello World") call_kwargs = mock_post.call_args[1] payload = call_kwargs["json"] # Validate chunks structure self.assertIn("chunks", payload) self.assertEqual(len(payload["chunks"]), 1) chunk = payload["chunks"][0] self.assertEqual(chunk["type"], "rich_text") text_element = ( chunk["elements"][0]["elements"][0] ) self.assertEqual(text_element["text"], "Hello World") print("✅ test_append_stream_success passed") @patch("requests.post") def test_stop_stream_success(self, mock_post): """Verify stopStream finalizes the stream.""" mock_response = MagicMock() mock_response.json.return_value = { "ok": True, "stream_ts": "1710000000.000100" } mock_post.return_value = mock_response stop_stream("C0123456789", "1710000000.000100") call_kwargs = mock_post.call_args[1] payload = call_kwargs["json"] self.assertEqual(payload["cursor_visibility"], "hidden") self.assertIn("chat.stopStream", mock_post.call_args[0][0]) print("✅ test_stop_stream_success passed") @patch("requests.post") def test_start_stream_failure(self, mock_post): """Verify error handling when startStream fails.""" mock_response = MagicMock() mock_response.json.return_value = { "ok": False, "error": "channel_not_found" } mock_post.return_value = mock_response with self.assertRaises(Exception) as context: start_stream("INVALID_CHANNEL") self.assertIn("channel_not_found", str(context.exception)) print("✅ test_start_stream_failure passed") if __name__ == "__main__": unittest.main(verbosity=2) ``` **Expected Test Output:** ``` test_append_stream_success ... ✅ test_append_stream_success passed ok test_start_stream_failure ... ✅ test_start_stream_failure passed ok test_start_stream_success ... ✅ test_start_stream_success passed ok test_stop_stream_success ... ✅ test_stop_stream_success passed ok ---------------------------------------------------------------------- Ran 4 tests in 0.005s OK ``` --- ## Sequential Block Update Flow ``` Client Slack API | | |-- chat.startStream ------>| |<-- stream_ts -------------| | | |-- chat.appendStream ----->| chunk 1: "Generating..." |<-- ok --------------------| | | |-- chat.appendStream ----->| chunk 2: "Here is the answer..." |<-- ok --------------------| | | |-- chat.appendStream ----->| chunk 3: "...complete." |<-- ok --------------------| | | |-- chat.stopStream ------->| |<-- ok --------------------| | [Stream finalized, cursor hidden] ``` --- ## Important Notes | Consideration | Details | |---------------|---------| | **Rate Limits** | Avoid sending chunks too rapidly; add small delays between `appendStream` calls | | **Error Handling** | Always call `stopStream` in a `finally` block to prevent zombie streams | | **Token Scope** | Bot token must have `chat:write` scope | | **`chunks` Type** | Only `rich_text` block type is currently supported for streaming | | **Thread Support** | Use `new_message_ts` in `appendStream` to stream into a thread |