# Streaming

> Real-time progress updates via Server-Sent Events.

**URL:** https://inference.sh/docs/api/sdk/streaming
**Last updated:** 2026-10-11

---

Real-time progress updates via Server-Sent Events.

This page is about following a task's output as it changes. For a two-way socket with a running task (live audio, a voice agent), see [Streaming and Live Functions](/docs/extend/streaming#calling-a-live-function).

---

## Basic Streaming

<div data-type="codegroup">

```python Python
from inferencesh import inference, TaskStatus

client = inference(api_key="inf_your_key")

for update in client.tasks.run({
    "app": "infsh/flux",
    "input": {"prompt": "A sunset"}
}, stream=True):
    print(f"Status: {TaskStatus(update['status']).name}")
    
    if update.get("logs"):
        print(f"Logs: {update['logs']}")
    
    if update["status"] == TaskStatus.COMPLETED:
        print(f"Output: {update['output']}")
```

```typescript JavaScript
const result = await client.run(
  { app: 'infsh/flux', input: { prompt: 'A sunset' } },
  {
    onUpdate: (update) => {
      console.log('Status:', update.status);
      if (update.logs) {
        console.log('Logs:', update.logs);
      }
    }
  }
);

console.log('Final output:', result.output);
```

```swift Swift
let task = try await client.tasks.run(
    ApiAppRunRequest(app: "infsh/flux", input: ["prompt": "A sunset"]),
    options: TaskRunOptions(
        onUpdate: { update in
            print("Status:", update.status.rawValue)
            if let logs = update.logs { print("Logs:", logs) }
        }
    )
)

print("Final output:", task.output)
```

</div>

---

## Stream Existing Task

<div data-type="codegroup">

```python Python
# Start task without waiting
task = client.tasks.run(params, wait=False)

# Stream updates later
with client.tasks.stream(task["id"]) as stream:
    for update in stream:
        print(f"Status: {update['status']}")
        if update["status"] == TaskStatus.COMPLETED:
            break

# Access final result
print(f"Final: {stream.result}")
```

```typescript JavaScript
// Start task
const task = await client.run(params);

// Track which fields changed
const result = await client.run(params, {
  onPartialUpdate: (update, fields) => {
    console.log('Changed fields:', fields);
    console.log('Update:', update);
  }
});
```

</div>

---

## Partial updates

The API may send incremental task updates as a `{data, fields}` envelope instead of a full task object. `fields` lists which keys changed in `data`.

<div data-type="codegroup">

```python Python
# TaskStream iterators receive the inner task object — no manual unwrapping
for update in client.tasks.run(params, stream=True):
    # Same shape as a GET /tasks/:id response
    print(update["status"], update.get("output"))
```

```typescript JavaScript
await client.run(params, {
  onPartialUpdate: (update, fields) => {
    console.log('Changed fields:', fields);
    console.log('Update:', update);
  }
});
```

</div>

The Python SDK unwraps `{data, fields}` lines automatically before yielding. Unlike the JavaScript `onPartialUpdate` callback, Python iterators do not expose the `fields` list — compare successive updates if you need to detect which keys changed.

The JavaScript SDK accumulates partial updates across the NDJSON stream. Fields present on the initial `POST /apps/run` response — such as `session_id` — are preserved in `onPartialUpdate` and the final resolved task even when later `{data, fields}` envelopes omit them.

---

## Stream completion and reconnect (JavaScript)

**@inferencesh/sdk ≥ v0.23.0:** `client.run()` (with `wait`, the default) and `client.tasks.watch()` in streaming mode reconnect the task's NDJSON stream after a network error or a stream that goes silent. On each reconnect the SDK fetches the task (`GET /tasks/:id`) before reading on, so a task that finished while the connection was down settles the wait. The API does not replay the lines a dropped stream missed: `onUpdate` and `onPartialUpdate` may skip intermediate states.

When the stream closes cleanly before the task is done (load balancer idle timeout, API deploy), the SDK fetches the task once and settles from it, or rejects when it is still not finished. `done` resolves when the task completes and rejects when it fails or is cancelled. A 4xx refusal rejects at once; other errors reject once the stream stops retrying. These waits always reconnect; with `stream: false` they poll instead (see [Reconnection Options](#reconnection-options)).

### Agent chat streams

- `agent.sendMessage()` and `agent.startStreaming()` reconnect the chat stream the same way and refetch the chat, with its messages, on each reconnect.
- `AgentChatProvider` reconnects its chat stream after a network drop mid-turn (for example Chromium `ERR_NETWORK_CHANGED`). Each reconnect refetches the chat and its messages, so a turn that finished meanwhile ends and `onTurnEnd` fires. `useAgentChat().connectionStatus` is `'connecting'` while a retry waits and `'error'` (with `error` set) once it gives up. When the stream ends on its own and the chat is still busy, the provider polls the chat until it is idle. See [React provider callbacks](/docs/api/sdk/polling#react-provider).

---

## Stream completion and GET reconciliation (Python)

`TaskStream` and `AsyncTaskStream` (from `client.tasks.run(..., stream=True)` and `client.tasks.stream()`) reconcile against `GET /tasks/:id` when the NDJSON stream ends without a terminal `COMPLETED`, `FAILED`, or `CANCELLED` event. This covers connection drops, heartbeat-only idle periods (default 30 seconds), and cases where the task finished server-side but the stream never delivered the final update.

```python
with client.tasks.run(params, stream=True) as stream:
    for update in stream:
        print(update["status"])

# Populated after reconciliation if the stream missed the terminal event
print(stream.result["output"])
```

| Outcome | Behavior |
|---------|----------|
| Task completed | Yields the reconciled task, sets `stream.result` |
| Task failed | Raises `RuntimeError` with the task `error` message |
| Task cancelled | Raises `RuntimeError("Task cancelled")` |
| Still running | Reconnects (when `auto_reconnect=True`) or exits |

Sync and async clients share this behavior.

`client.tasks.wait_for_completion()` uses the same stream-and-reconcile path internally. If the NDJSON stream ends without a terminal event, the client performs a final `GET /tasks/:id` before returning or raising.

Pass an optional `timeout` (seconds) to cap how long the call waits. When the deadline is reached, the client still attempts that final GET — a task that finished while the stream was idle is returned even if the timeout has elapsed. If the task is still non-terminal after the GET, `TimeoutError` is raised. Failed or cancelled tasks raise `RuntimeError` immediately from the stream or from GET reconciliation.

```python
try:
    result = client.tasks.wait_for_completion(task_id, timeout=300)
except TimeoutError:
    print("Task still running after 5 minutes")
except RuntimeError as e:
    print(f"Task failed or cancelled: {e}")
```

The async client accepts the same `timeout` keyword: `await client.tasks.wait_for_completion(task_id, timeout=300)`.

---

## Combined Callbacks (JavaScript)

```typescript
await client.run(params, {
  onUpdate: (update) => {
    console.log(`Status: ${update.status}`);
  },
  onPartialUpdate: (update, fields) => {
    if (fields.includes('logs')) {
      console.log('New logs:', update.logs);
    }
  }
});
```

---

## Async Streaming (Python)

```python
from inferencesh import async_inference, TaskStatus

async def main():
    client = async_inference(api_key="inf_your_key")

    # Stream while running a new task
    async for update in await client.tasks.run(params, stream=True):
        print(f"Status: {update['status']}")

    # Or stream an existing task
    async with client.tasks.stream(task_id) as stream:
        async for update in stream:
            if update["status"] == TaskStatus.COMPLETED:
                print(update["output"])
                break
```

---

## Reconnection Options

<div data-type="codegroup">

```python Python
# Tune SSE performance
client = inference(
    api_key="inf_your_key",
    sse_chunk_size=8192,
    sse_mode="iter_lines"
)

# Or via environment variables
# export INFERENCE_SSE_READ_BYTES=8192
# export INFERENCE_SSE_MODE=iter_lines

# Per-stream reconnect tuning (also on client.tasks.stream())
with client.tasks.run(
    params,
    stream=True,
    auto_reconnect=True,
    max_reconnects=5,
    reconnect_delay_ms=1000,
) as stream:
    for update in stream:
        print(update["status"])
```

```typescript JavaScript
// Polling mode retry tuning (when stream: false on client or per-run)
await client.run(params, {
  stream: false,
  maxReconnects: 5,
  pollIntervalMs: 2000,
  onUpdate: (update) => console.log(update)
});

// Your own NDJSON stream with reconnect (v0.23.0+)
import { StreamableManager, type StreamReconnectOptions, type Task } from '@inferencesh/sdk';

const policy: StreamReconnectOptions = {
  maxAttempts: 10,
  baseDelayMs: 1000,
  maxDelayMs: 30_000,
  idleTimeoutMs: 30_000, // 0 turns idle reconnect off; the API sends a heartbeat every 10 s
};

const manager = new StreamableManager<Task>({
  request: (init) => client.http.fetch(`/tasks/${taskId}/stream`, init),
  reconnect: policy, // or `true` for these defaults
  onRetry: (error, attempt, delayMs) => console.warn(`retry ${attempt} in ${delayMs} ms`, error.message),
  onReconnect: async () => {
    // Lines sent while the stream was down are lost: fetch the current state
    const { data } = await client.tasks.get(taskId);
    console.log(data.status);
  },
  onData: (update) => console.log(update.status),
});
await manager.start();
```

</div>

`client.run()` streams task progress over NDJSON (`StreamableManager`) by default and reconnects on its own (see [Stream completion and reconnect](#stream-completion-and-reconnect-javascript)); it does not take `autoReconnect`. **@inferencesh/sdk ≥ v0.23.0** gives a `StreamableManager` you create yourself an opt-in `reconnect`: `true`, or `StreamReconnectOptions` (`maxAttempts`, `baseDelayMs`, `maxDelayMs`, `idleTimeoutMs`; defaults 10, 1000, 30000, 30000). Without it, a failure ends the stream. Retries use jittered exponential backoff, and any line received, heartbeats included, resets the failure count. A stream silent for `idleTimeoutMs` is reconnected. In a browser, going `online` or the tab becoming visible retries at once. A 4xx other than 408 and 429 is not retried, and a clean end of the stream is not retried. With `reconnect`, `onError` fires only when the manager gives up. Use `StreamManager` (see below) for EventSource streams.

---

## SSE EventSource streams (JavaScript)

Namespace `.stream()` helpers return a browser `EventSource` for SSE endpoints — for example `client.tasks.stream(taskId)`, `client.chats.stream(chatId)`, `client.flows.stream(flowId)`, and `client.flowRuns.stream(flowRunId)`.

On the server, **@inferencesh/sdk ≥ v0.9.0** uses `eventsource` 5.x and requires **Node.js 22.12+** — see [JavaScript SDK — Installation](/docs/api/sdk-javascript#installation).

Failed **initial** stream connections (HTTP 401/403, OTP, expired tokens, network errors before bytes arrive) are routed through the client-level `onError` handler — the same path as REST and `HttpClient.fetch()`. That includes SSE handshakes on `.stream()` helpers and NDJSON opened by the SDK (`tasks.run()` / `tasks.watch()`, agent chat). **@inferencesh/sdk ≥ v0.21.0** passes a third argument, `FailedRequest` (`{ token }`), with the bearer token the failed attempt used so you can tell stale in-flight requests from ones sent after a refresh.

Return the result of `retry()` from your handler to reconnect after refreshing credentials (or return a `Response` for `fetch()` / EventSource handshakes):

```typescript
import { inference, InferenceError } from '@inferencesh/sdk';

const client = inference({
  apiKey: 'inf_your_key',
  onError: async (error, retry, request) => {
    if (error instanceof InferenceError && error.statusCode === 401) {
      await refreshToken();
      return retry();
    }
    throw error;
  },
});

const source = await client.tasks.stream('task_abc123');
```

For lower-level control, `StreamManager` from `@inferencesh/sdk` wraps `EventSource` with `autoReconnect`, `maxReconnects`, and `reconnectDelayMs` for drops **after** the stream is open:

```typescript
import { createHttpClient, StreamManager } from '@inferencesh/sdk';

const http = createHttpClient({ apiKey: 'inf_your_key' });
const manager = new StreamManager({
  createEventSource: () => http.createEventSource('/tasks/task_abc123/stream'),
  autoReconnect: true,
  maxReconnects: 5,
  reconnectDelayMs: 1000,
  onData: (update) => console.log(update.status),
});
manager.start();
```

| Mechanism | When it applies |
|-----------|-----------------|
| `onError` (client config) | REST, `HttpClient.fetch()`, refused or failed **initial** NDJSON/SSE connections (auth, OTP, requirements, network before open) |
| `autoReconnect` (`StreamManager`) | Connection lost after the EventSource is open |

For custom NDJSON with dynamic tokens, pass a `StreamRequest` to `StreamableManager` (or `streamable()` / `streamableRaw()`) — for example `(init) => http.fetch('/tasks/{id}/stream', init)` — instead of baking URL and headers from `getStreamableConfig()`. See [JavaScript SDK — Error Handling](/docs/api/sdk-javascript#error-handling).

→ [Error Handling](/docs/api/sdk-javascript#error-handling) — `InferenceError` shape and RFC 9457 parsing

---

## Status Constants

<div data-type="codegroup">

```python Python
from inferencesh import TaskStatus

TaskStatus.RECEIVED    # 1 - Request received
TaskStatus.QUEUED      # 2 - Waiting in queue
TaskStatus.DISPATCHED  # 3 - Worker assigned (legacy alias: SCHEDULED)
TaskStatus.PREPARING   # 4 - Preparing
TaskStatus.SERVING     # 5 - Serving
TaskStatus.SETTING_UP  # 6 - Setting up
TaskStatus.RUNNING     # 7 - Executing
TaskStatus.CANCELLING  # 8 - Cancelling
TaskStatus.UPLOADING   # 9 - Uploading results
TaskStatus.COMPLETED   # 10 - Done
TaskStatus.FAILED      # 11 - Error
TaskStatus.CANCELLED   # 12 - Cancelled
```

```typescript JavaScript
import {
  TaskStatusReceived,    // 1
  TaskStatusQueued,      // 2
  TaskStatusDispatched,  // 3
  TaskStatusPreparing,   // 4
  TaskStatusServing,     // 5
  TaskStatusSettingUp,   // 6
  TaskStatusRunning,     // 7
  TaskStatusCancelling,  // 8
  TaskStatusUploading,   // 9
  TaskStatusCompleted,   // 10
  TaskStatusFailed,      // 11
  TaskStatusCancelled    // 12
} from '@inferencesh/sdk';
```

</div>