Async operations

Some operations on Stream's API take longer than a single HTTP response can wait for. Hard-deleting channels, deleting users in bulk, exporting users or calls, and similar batch jobs return a task_id immediately and run in the background. To get the result, poll the task status endpoint.

Task statuses

A task moves through the following statuses:

• pending: the task is queued and not running yet
• running: the task is running and not completed yet
• completed: the task finished successfully
• failed: the task failed during its execution

The task status response carries the current status, a result payload whose shape depends on the task, and an error with failure details when the task failed.

Waiting on a task

The unified SDKs ship a helper that polls for you and surfaces the outcome as either a typed result (when the task completes) or a typed exception (when it fails or the wait elapses).

from getstream.exceptions import StreamTaskException, StreamTransportException

response = client.delete_channels(cids=["messaging:c1", "messaging:c2"], hard_delete=True)
task_id = response.task_id

try:
    result = client.wait_for_task(task_id)
    print("task completed:", result)
except StreamTaskException as e:
    print(f"task {e.task_id} failed: {e.description}")
except StreamTransportException as e:
    if e.error_type == "timeout":
        # wait elapsed; task may still be running on the server
        ...

begin
  result = client.wait_for_task(task_id)
  # task completed
rescue GetStreamRuby::TaskError => e
  warn "task #{e.task_id} failed: #{e.description}"
rescue GetStreamRuby::TransportError => e
  warn 'wait elapsed; task may still be running' if e.error_type == 'timeout'
end

use GetStream\Exceptions\StreamTaskException;
use GetStream\Exceptions\StreamTransportException;

try {
    $result = $client->waitForTask($taskId);
    // task completed
} catch (StreamTaskException $e) {
    error_log("task {$e->getTaskId()} failed: {$e->getDescription()}");
} catch (StreamTransportException $e) {
    if ($e->getErrorType() === 'timeout') {
        // wait elapsed; task may still be running on the server
    }
}

import (
    "context"
    "errors"
    "github.com/GetStream/getstream-go/v4"
)

resp, err := client.Chat().DeleteChannels(ctx, &getstream.DeleteChannelsRequest{
    Cids:       []string{"messaging:c1", "messaging:c2"},
    HardDelete: getstream.PtrTo(true),
})
if err != nil {
    return err
}

taskResp, err := getstream.WaitForTask(ctx, client, *resp.Data.TaskID)
if err != nil {
    var streamErr *getstream.StreamError
    if errors.As(err, &streamErr) {
        if errors.Is(err, getstream.ErrTaskFailed) {
            // task ended with status: failed; streamErr.Task carries the details
        } else if errors.Is(err, getstream.ErrTransport) && streamErr.ErrorType == "timeout" {
            // wait elapsed; task may still be running on the server
        }
    }
}

using GetStream;

try
{
    var result = await client.WaitForTaskAsync(taskId);
    // result.Status == "completed"
}
catch (GetStreamTaskException ex)
{
    // task ended with status: failed
    Console.WriteLine($"task {ex.TaskId} failed: {ex.Description}");
}
catch (GetStreamTransportException ex) when (ex.ErrorType == "timeout")
{
    // wait elapsed; task may still be running on the server
}

import io.getstream.exceptions.StreamTaskException;
import io.getstream.exceptions.StreamTransportException;

try {
    var result = client.waitForTask(taskId);
    // task completed
} catch (StreamTaskException e) {
    System.err.println("task " + e.getTaskId() + " failed: " + e.getDescription());
} catch (StreamTransportException e) {
    if ("timeout".equals(e.getErrorType())) {
        // wait elapsed; task may still be running on the server
    }
}

The task and transport exceptions are part of the SDK exception hierarchy documented in Error handling.

Behavior

Defaults

Override either knob if your task is expected to run longer, or if you want a tighter loop.

client.wait_for_task(task_id, poll_interval=5.0, timeout=600.0)

client.wait_for_task(task_id, poll_interval: 5, timeout: 600)

$client->waitForTask($taskId, 5, 600);

import "time"

getstream.WaitForTask(ctx, client, taskID,
    getstream.WithWaitForTaskPollInterval(5*time.Second),
    getstream.WithWaitForTaskTimeout(10*time.Minute),
)

await client.WaitForTaskAsync(taskId, TimeSpan.FromSeconds(5), TimeSpan.FromMinutes(10));

client.waitForTask(taskId, Duration.ofSeconds(5), Duration.ofMinutes(10));

The async variant in your SDK (for example Python's AsyncStream.wait_for_task or .NET's WaitForTaskAsync) is non-blocking and accepts the same parameters.

Polling manually

If you need a custom polling loop (back-off, progress logging, external cancellation), call getTask yourself. The helper is a convenience over the same endpoint.

const response = await client.exportUsers({
  user_ids: ["<user id1>", "<user id2>"],
});

// poll this endpoint until the status is completed or failed
const taskResponse = await client.getTask({ id: response.task_id });

console.log(taskResponse.status === "completed");

import time

while True:
    response = client.get_task(task_id)
    if response.data.status in ("completed", "failed"):
        break
    time.sleep(1.0)

for {
    resp, err := client.GetTask(ctx, taskID, &getstream.GetTaskRequest{})
    if err != nil { return err }
    if resp.Data.Status == "completed" || resp.Data.Status == "failed" {
        break
    }
    time.Sleep(time.Second)
}

var taskStatus = client.getTask(taskID).execute();

if (taskStatus.getData().getStatus().equals("completed")) {
    System.out.println(taskStatus.getData().getResult());
}

# When an operation is async, a task_id is included in the API response.
# Use it to monitor the task; when finished, the status is completed.
curl -X GET https://chat.stream-io-api.com/api/v2/tasks/${TASK_ID}?api_key=${API_KEY} \
    -H "Authorization: ${TOKEN}" \
    -H "stream-auth-type: jwt"