> ## Documentation Index
> Fetch the complete documentation index at: https://developer.watson-orchestrate.ibm.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Handling async flows in API-driven chat

> How to detect when an agent starts an async flow and poll for messages while the flow runs.

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.

<Note>
  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.
</Note>

## How async flows work

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

1. The SSE stream emits a `flow.slot.listen` event containing the `thread_id`.
2. The stream ends — no further `message.delta` events arrive on this connection.
3. The flow runs asynchronously. The thread status changes to `async_wait`.
4. When the flow needs user input (for example, a form or confirmation), the thread status changes to `async_slot_request`.
5. 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.

```python theme={null}
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.

```python theme={null}
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.

```python theme={null}
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

```python theme={null}
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.