Handling async flows in API-driven chat

When you drive chat entirely through the API using POST /v1/orchestrate/runs/stream, most agent responses arrive synchronously as SSE events. However, when the agent triggers an async flow (a tool flow that runs asynchronously), the stream pauses and the agent sends a holding message. Your application must detect this condition and poll the message thread until the flow completes and the agent resumes. This guide explains how to detect the async flow condition and how to poll for messages correctly.

This pattern applies only to applications that consume the watsonx Orchestrate chat API directly. If you use the embedded chat widget, the widget handles async flow polling automatically.

How async flows work

When the agent starts an async flow during a streaming run:

The SSE stream emits a flow.slot.listen event containing the thread_id.
The stream ends — no further message.delta events arrive on this connection.
The flow runs asynchronously. The thread status changes to async_wait.
When the flow needs user input (for example, a form or confirmation), the thread status changes to async_slot_request.
When the flow completes, new messages are appended to the thread and the thread status returns to ready.

Your application is responsible for polling the thread’s messages between steps 2 and 5.

Step 1 — Start a streaming run and capture the thread ID

Send a message to POST /v1/orchestrate/runs/stream. Include the thread_id from the previous turn if you are continuing a conversation; omit it for a new conversation.

import httpx
import json

BASE_URL = "https://api.<your-saas-hostname>/instances/<tenant_id>"
HEADERS = {
    "Authorization": "Bearer <your_token>",
    "Content-Type": "application/json",
}

def stream_run(thread_id: str | None, user_message: str) -> dict:
    """
    Start a streaming run and return the final event data.
    Returns a dict with keys: thread_id, status, flow_started.
    """
    body = {
        "message": {"role": "user", "content": [{"type": "text", "text": user_message}]},
    }
    if thread_id:
        body["thread_id"] = thread_id

    flow_started = False
    result_thread_id = thread_id

    with httpx.stream(
        "POST",
        f"{BASE_URL}/v1/orchestrate/runs/stream",
        headers=HEADERS,
        json=body,
        timeout=60,
    ) as response:
        response.raise_for_status()
        for line in response.iter_lines():
            if not line.startswith("data: "):
                continue
            payload = line.removeprefix("data: ").strip()
            if payload in ("", "[DONE]"):
                continue
            event = json.loads(payload)
            event_type = event.get("event")
            data = event.get("data", {})

            # Always capture the thread_id from any event that carries it
            if data.get("thread_id"):
                result_thread_id = data["thread_id"]

            if event_type == "flow.slot.listen":
                # Async flow has started — the stream will end without further message content
                flow_started = True

            elif event_type == "message.delta":
                # Normal text response — print or accumulate
                for chunk in data.get("delta", {}).get("content", []):
                    print(chunk.get("text", ""), end="", flush=True)

            elif event_type in ("message.completed", "done"):
                break

    print()  # newline after streamed output
    return {"thread_id": result_thread_id, "flow_started": flow_started}

Step 2 — Detect the async flow condition

Check the return value of stream_run. When flow_started is True, the agent has handed off to a flow and you must poll.

result = stream_run(thread_id=None, user_message="Submit a PTO request for next Friday")

thread_id = result["thread_id"]
flow_started = result["flow_started"]

if flow_started:
    print(f"Flow started on thread {thread_id}. Polling for messages...")
    poll_for_messages(thread_id)

Step 3 — Poll for new messages

Poll GET /v1/orchestrate/threads/{thread_id}/messages at a regular interval. Compare the message list with what you already have and display any new messages as they appear. The flow is still running while the thread status is async_wait. When the flow needs user input, the status changes to async_slot_request — this is your cue to prompt the user. When the status returns to ready, the flow has completed.

import time

def get_thread_messages(thread_id: str) -> list:
    response = httpx.get(
        f"{BASE_URL}/v1/orchestrate/threads/{thread_id}/messages",
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()
    return response.json()

def get_thread_status(thread_id: str) -> str:
    response = httpx.get(
        f"{BASE_URL}/v1/orchestrate/threads/{thread_id}",
        headers=HEADERS,
        timeout=30,
    )
    response.raise_for_status()
    return response.json().get("status", "ready")

def poll_for_messages(thread_id: str, poll_interval: float = 2.0, timeout: float = 300.0):
    """
    Poll the thread for new messages until the flow completes.
    Displays each new assistant message as it appears.
    """
    seen_ids = set()
    elapsed = 0.0

    while elapsed < timeout:
        status = get_thread_status(thread_id)
        messages = get_thread_messages(thread_id)

        # Display any new messages
        for msg in messages:
            if msg["id"] in seen_ids:
                continue
            seen_ids.add(msg["id"])

            if msg.get("role") == "assistant":
                for item in msg.get("content", []):
                    if item.get("response_type") == "text":
                        print(f"Agent: {item['text']}")

        if status == "async_slot_request":
            # Flow is waiting for user input
            user_reply = input("Your response: ")
            # Send the user's reply back on the same thread
            result = stream_run(thread_id=thread_id, user_message=user_reply)
            if not result["flow_started"]:
                # Flow has completed; agent responded normally
                break

        elif status == "ready":
            # Flow has completed
            break

        time.sleep(poll_interval)
        elapsed += poll_interval

    if elapsed >= timeout:
        print("Polling timed out — the flow may still be running.")

Putting it together

def chat(thread_id: str | None = None):
    user_message = input("You: ")
    result = stream_run(thread_id=thread_id, user_message=user_message)

    thread_id = result["thread_id"]

    if result["flow_started"]:
        poll_for_messages(thread_id)

    return thread_id


# Start a conversation
thread_id = chat()

# Continue the same conversation
thread_id = chat(thread_id=thread_id)

Thread status reference

Status	Meaning
`ready`	No async operation in progress. Normal streaming turns work as expected.
`async_wait`	An async flow is running. The thread is not accepting new user messages. Poll for messages.
`async_slot_request`	The flow is waiting for user input. Send the user’s reply via `POST /v1/orchestrate/runs/stream` with the same `thread_id`.

SSE event reference

Event type	When it fires	What to do
`message.delta`	Assistant is streaming a text response	Accumulate and display the content chunks
`flow.slot.listen`	Agent has started an async flow	Record the `thread_id` and begin polling
`message.completed`	A complete message has been delivered	End of normal turn
`done`	Stream session ended	End of streaming connection
`error`	A turn-level error occurred	Surface the error to the user

Notes

The flow.slot.listen event is the definitive signal that async polling is required. Do not rely on the absence of further message.delta events as the detection mechanism.
The WebSocket-based notification API (equivalent to how the native chat UI receives live flow updates) is not yet available for external API consumers. Polling /messages is the supported approach.
Poll intervals of 1–3 seconds are appropriate for most flows. Avoid polling faster than once per second.
Always use the thread_id returned in the SSE events — do not cache or infer it from elsewhere.

Release Notes

Get Started

Build

Deploy

Analyze

Developer experience

Legal notices

How async flows work

Step 1 — Start a streaming run and capture the thread ID

Step 2 — Detect the async flow condition

Step 3 — Poll for new messages

Putting it together

Thread status reference

SSE event reference

Notes

​How async flows work

​Step 1 — Start a streaming run and capture the thread ID

​Step 2 — Detect the async flow condition

​Step 3 — Poll for new messages

​Putting it together

​Thread status reference

​SSE event reference

​Notes

How async flows work

Step 1 — Start a streaming run and capture the thread ID

Step 2 — Detect the async flow condition

Step 3 — Poll for new messages

Putting it together

Thread status reference

SSE event reference

Notes