Streaming Patterns

The /automate and /research endpoints always stream via Server-Sent Events (SSE). This guide covers how to handle that stream correctly across TypeScript and Python, and in different runtime environments.

---

Why these endpoints stream

Automation and research tasks take time: seconds to minutes. SSE lets you show real-time progress to users, handle partial results, and detect failures early rather than waiting for a timeout.

There is no non-streaming mode for /automate or /research. Every call streams.

---

TypeScript: Node.js

The TypeScript SDK returns an async iterable stream. Iterate with for await. The SDK models the stream as a discriminated union, so switching on event.event narrows event.data to the correct type — no casts needed:

import Tabstack from '@tabstack/sdk'

const client = new Tabstack({ apiKey: process.env.TABSTACK_API_KEY })

const stream = await client.agent.research({
  query: 'What are the current pricing models for cloud browser APIs?',
  mode: 'balanced'
})

for await (const event of stream) {
  switch (event.event) {
    case 'iteration:start':
      process.stdout.write(
        `\r[iteration ${event.data.iteration}/${event.data.maxIterations}]`,
      )
      break
    case 'complete':
      console.log('\n\n' + event.data.report)
      break
    case 'error':
      console.error('\nError:', event.data.error?.message ?? 'unknown error')
      break
  }
}

/research emits a rich set of progress events — planning:*, iteration:*, searching:*, writing:*, and (on balanced mode) additional prefetching:*, analyzing:*, following:*, evaluating:*, outlining:*, judging:* events. Switch on the ones you want to display; ignore the rest. The error event’s data.error is an object with message, name, and optional stack.

---

TypeScript: Edge Runtimes (Vercel Edge, Cloudflare Workers)

Edge runtimes support the SDK natively. The SDK uses fetch under the hood, which is available globally in edge environments:

import Tabstack from '@tabstack/sdk'

export const runtime = 'edge'

export async function GET() {
  const client = new Tabstack({ apiKey: process.env.TABSTACK_API_KEY })

  const stream = await client.agent.research({
    query: 'What are the latest LLM benchmarks?',
    mode: 'fast'
  })

  // Forward the SSE stream to the browser
  const encoder = new TextEncoder()
  const readable = new ReadableStream({
    async start(controller) {
      for await (const event of stream) {
        const chunk = `event: ${event.event}\ndata: ${JSON.stringify(event.data)}\n\n`
        controller.enqueue(encoder.encode(chunk))

        if (event.event === 'complete' || event.event === 'error') {
          controller.close()
          break
        }
      }
    }
  })

  return new Response(readable, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
    }
  })
}

---

TypeScript: Bun

Bun supports the SDK natively. Same pattern as Node.js:

import Tabstack from '@tabstack/sdk'

const client = new Tabstack({ apiKey: Bun.env.TABSTACK_API_KEY })

const stream = await client.agent.automate({
  task: 'Extract the top 10 repositories from GitHub trending',
  url: 'https://github.com/trending',
  guardrails: 'Browse and extract only.'
})

for await (const event of stream) {
  if (event.event === 'complete') {
    console.log(JSON.stringify(event.data, null, 2))
  }
}

---

Python: Synchronous

The primary Python pattern iterates the stream directly. The SDK handles SSE framing (event: and data: lines) internally — you don’t need to parse raw lines. event.data is a typed model; access fields as attributes, not with .get():

import os
from tabstack import Tabstack

client = Tabstack(api_key=os.environ.get("TABSTACK_API_KEY"))

for event in client.agent.research(
    query="What are the current pricing models for cloud browser APIs?",
    mode="balanced",
):
    if event.event == "complete":
        print(event.data.report)
    elif event.event == "error":
        message = getattr(event.data.error, "message", None) or "unknown error"
        print("Error:", message)

The SDK also exposes client.agent.with_streaming_response.research(...) as a context manager for cases where you need the raw httpx.Response (custom timeout handling, manual line iteration). The direct iterator above is the recommended pattern for nearly all consumers.

---

Python: Async

import os
import asyncio
from tabstack import AsyncTabstack


async def run_research():
    async with AsyncTabstack(api_key=os.environ.get("TABSTACK_API_KEY")) as client:
        stream = await client.agent.research(
            query="What are the latest GPU benchmarks?",
            mode="fast",
        )
        async for event in stream:
            if event.event == "complete":
                return event.data.report
            if event.event == "error":
                message = getattr(event.data.error, "message", None) or "unknown error"
                raise RuntimeError(message)


result = asyncio.run(run_research())

---

Always handle the error event

Across all languages: always handle error events. An SSE error is a task-level failure that arrives in the stream, not an HTTP error. If you only handle complete, failures will silently produce no output.

// Minimal correct handler (for /research)
for await (const event of stream) {
  if (event.event === 'complete') {
    return event.data.report
  }
  if (event.event === 'error') {
    throw new Error(event.data.error?.message ?? 'unknown error')
  }
}