Streaming

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.

Basic Streaming

1from inferencesh import inference, TaskStatus23client = inference(api_key="inf_your_key")45for update in client.tasks.run({6    "app": "infsh/flux",7    "input": {"prompt": "A sunset"}8}, stream=True):9    print(f"Status: {TaskStatus(update['status']).name}")10    11    if update.get("logs"):12        print(f"Logs: {update['logs']}")13    14    if update["status"] == TaskStatus.COMPLETED:15        print(f"Output: {update['output']}")

Stream Existing Task

1# Start task without waiting2task = client.tasks.run(params, wait=False)34# Stream updates later5with client.tasks.stream(task["id"]) as stream:6    for update in stream:7        print(f"Status: {update['status']}")8        if update["status"] == TaskStatus.COMPLETED:9            break1011# Access final result12print(f"Final: {stream.result}")

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.

1# TaskStream iterators receive the inner task object — no manual unwrapping2for update in client.tasks.run(params, stream=True):3    # Same shape as a GET /tasks/:id response4    print(update["status"], update.get("output"))

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).

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.

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.

1with client.tasks.run(params, stream=True) as stream:2    for update in stream:3        print(update["status"])45# Populated after reconciliation if the stream missed the terminal event6print(stream.result["output"])

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.

1try:2    result = client.tasks.wait_for_completion(task_id, timeout=300)3except TimeoutError:4    print("Task still running after 5 minutes")5except RuntimeError as e:6    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)

1await client.run(params, {2  onUpdate: (update) => {3    console.log(`Status: ${update.status}`);4  },5  onPartialUpdate: (update, fields) => {6    if (fields.includes('logs')) {7      console.log('New logs:', update.logs);8    }9  }10});

Async Streaming (Python)

1from inferencesh import async_inference, TaskStatus23async def main():4    client = async_inference(api_key="inf_your_key")56    # Stream while running a new task7    async for update in await client.tasks.run(params, stream=True):8        print(f"Status: {update['status']}")910    # Or stream an existing task11    async with client.tasks.stream(task_id) as stream:12        async for update in stream:13            if update["status"] == TaskStatus.COMPLETED:14                print(update["output"])15                break

Reconnection Options

1# Tune SSE performance2client = inference(3    api_key="inf_your_key",4    sse_chunk_size=8192,5    sse_mode="iter_lines"6)78# Or via environment variables9# export INFERENCE_SSE_READ_BYTES=819210# export INFERENCE_SSE_MODE=iter_lines1112# Per-stream reconnect tuning (also on client.tasks.stream())13with client.tasks.run(14    params,15    stream=True,16    auto_reconnect=True,17    max_reconnects=5,18    reconnect_delay_ms=1000,19) as stream:20    for update in stream:21        print(update["status"])

client.run() streams task progress over NDJSON (StreamableManager) by default and reconnects on its own (see Stream completion and reconnect); 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.

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):

1import { inference, InferenceError } from '@inferencesh/sdk';23const client = inference({4  apiKey: 'inf_your_key',5  onError: async (error, retry, request) => {6    if (error instanceof InferenceError && error.statusCode === 401) {7      await refreshToken();8      return retry();9    }10    throw error;11  },12});1314const 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:

1import { createHttpClient, StreamManager } from '@inferencesh/sdk';23const http = createHttpClient({ apiKey: 'inf_your_key' });4const manager = new StreamManager({5  createEventSource: () => http.createEventSource('/tasks/task_abc123/stream'),6  autoReconnect: true,7  maxReconnects: 5,8  reconnectDelayMs: 1000,9  onData: (update) => console.log(update.status),10});11manager.start();

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.

→ Error Handling — InferenceError shape and RFC 9457 parsing

Status Constants

1from inferencesh import TaskStatus23TaskStatus.RECEIVED    # 1 - Request received4TaskStatus.QUEUED      # 2 - Waiting in queue5TaskStatus.DISPATCHED  # 3 - Worker assigned (legacy alias: SCHEDULED)6TaskStatus.PREPARING   # 4 - Preparing7TaskStatus.SERVING     # 5 - Serving8TaskStatus.SETTING_UP  # 6 - Setting up9TaskStatus.RUNNING     # 7 - Executing10TaskStatus.CANCELLING  # 8 - Cancelling11TaskStatus.UPLOADING   # 9 - Uploading results12TaskStatus.COMPLETED   # 10 - Done13TaskStatus.FAILED      # 11 - Error14TaskStatus.CANCELLED   # 12 - Cancelled