Orchestrator

The Orchestrator module enables your application to run a complete AI workflow. Unlike an Agent, which interacts with a single AI, an Orchestrator manages multiple steps, such as invoking multiple agents, and following branching logic, all configured in the ART Orchestrator Builder.

Prerequisites

Before using the Orchestrator module, ensure the following:

The ADK is installed, authenticated, and connected. If you haven't completed this setup yet, see the Installation guide.
An Orchestrator has been created and deployed using the ART Orchestrator Builder.
You have the orchestratorId of the deployed Orchestrator.

Workflow

Your application sends a request to an Orchestrator, which executes the workflow and streams events in real time as each step progresses. Each Orchestrator communicates through its own dedicated channel.

Unlike the Agent module, which provides typed events and Run objects, the Orchestrator module uses a flexible, event-based communication model. You send named events and receive raw objects {event, content}, giving you full control over the data exchanged during workflow execution.

To simplify working with workflows, the ADK provides two core classes that you'll use throughout the Orchestrator module:

Class	Description
`Orchestrator`	Represents a single orchestrator identified by its `orchestratorId`. Reuse the same `Orchestrator` instance for multiple workflow executions. The ADK automatically manages the underlying communication channel.
`OrchestratorThread`	Represents a single workflow execution. Each thread has a unique `threadId`, which is automatically included with every event to associate it with the correct workflow execution.

Quick start

The following example demonstrates the basic workflow: obtaining an orchestrator, creating a thread, registering an event listener, and sending a user_input event to start the workflow:

// 1. Get the orchestrator and open a thread.
let orchestrator: Orchestrator = adk.orchestrator("your-orchestrator-id")

// 2. Start a thread.
let thread: OrchestratorThread = await orchestrator.thread()

// 3. Listen for the events the workflow sends back.
await thread.listen { (data: [String: Any]) in
    guard
        let content = data["content"] as? [String: Any],
        let type = content["type"] as? String,
        type == "agent_general_response",
        let message = content["message"] as? String
    else {
        return
    }

    print("Answer: \(message)")
}

// 4. Send the user's request.
await thread.push(
    event: "user_input",
    data: [
        "user_input": "Plan a 3-day trip to Goa"
    ]
)

The following sections explain each step in more detail, including how to handle workflow events and respond to human_input_request events.

1. Connect to an orchestrator

Get an adk.orchestrator(...) instance from the connected Adk instance. The channel subscription is established lazily — it is established automatically the first time you create an OrchestratorThread by calling orchestrator.thread():

let orchestrator: Orchestrator = adk.orchestrator("your-orchestrator-id")

2. Start a thread

An OrchestratorThread represents a single workflow execution. Its threadId is automatically attached to every outbound message, ensuring that responses are routed to the correct thread:

let thread: OrchestratorThread = await orchestrator.thread()

print(thread.threadId)

thread() is asynchronous because it establishes the underlying channel subscription the first time it is called. To resume a previous workflow execution, pass the existing threadId:

3. Listen to the thread

thread.listen registers a callback that receives every event emitted for the thread. Each event is delivered as a raw map containing two fields: event and content. Determine the event type from content['type']:

await thread.listen { (data: [String: Any]) in
    guard let content = data["content"] as? [String: Any] else {
        print("Event: \(data["event"] ?? "unknown")")
        return
    }

    let type = content["type"] as? String

    switch type {
    case "agent_general_response":
        print("Answer: \(content["message"] ?? "")")

    case "human_input_request":
        print("Workflow asks: \(content["prompt"] ?? "")")

    case "agent_error_response":
        print("Error: \(content["message"] ?? "")")

    case "agent_wait_response":
        print("Waiting on: \(content["waiting_for_agent_id"] ?? "")")

    case "planner_correction_request":
        print("Revising plan: \(content["reason"] ?? "")")

    default:
        print("Event: \(data["event"] ?? "unknown")")
    }
}

Listening to a single event

To listen for a single event type, use bind(). Any buffered events with the specified name are delivered first, in the order they were received, followed by future events of the same type:

thread.bind(
    event: "status_update"
) { content in
    print("status: \(content)")
}

// Stop listening to that event:
thread.remove(event: "status_update")

bind() differs from listen() in two ways:

It invokes the callback only for the specified event type.
It passes the event's content directly, rather than the raw {event, content} map.

Use bind() when you only care about a specific workflow event. Use listen() when you need to observe the complete event stream.

4. Send the user's request

Send the user's request as a user_input event. The thread automatically stamps its thread_id on the frame — you only provide the payload:

await thread.push(
    event: "user_input",
    data: [
        "user_input": "Plan a 3-day trip to Goa"
    ]
)

The payload is any JSON-serializable map, so additional workflow inputs are just extra keys:

await thread.push(
    event: "user_input",
    data: [
        "user_input": "Plan a trip",
        "preferences": [
            "budget": 50_000,
            "days": 3
        ]
    ]
)

Parameters

Parameter	Type	Description
`event`	`String`	The event name, for example `user_input` or `user_reply`.
`data`	`[String: Any]`	The event payload. Its structure is defined by your workflow.
`options`	`PushConfig?`	Optional delivery configuration. The `OrchestratorThread` automatically sets the `threadId` for every outbound message.

5. Human-in-the-Loop (HITL)

Like the Agent, the Orchestrator supports Human-in-the-Loop (HITL) interactions. A workflow can pause at any point to request additional input from the user. When this happens, the ADK injects a reply function into the event's content. Call this function with the user's response to resume the workflow execution:

 thread.listen { (data: [String: Any]) in
    guard let content = data["content"] as? [String: Any] else {
        return
    }

    if let prompt = content["prompt"] as? String,
       let reply = content["reply"] as? ([String: Any]) -> Void {

        print("Workflow asks: \(prompt)")

        // Collect the answer from your UI, then call reply.
        reply([
            "user_input": "Budget 50k, December 20-25"
        ])
        return
    }

    // …handle other events…
}

6. Clean up

When a workflow execution is complete, dispose of the OrchestratorThread. This unregisters the thread from the underlying channel and detaches all associated listeners:

thread.dispose()

dispose() is idempotent. It does not close the underlying channel subscription or the WebSocket connection, so other OrchestratorThread instances continue to operate normally. After a thread has been disposed, calling push() or listen() on that thread throws a StateError. Use thread.isDisposed to determine whether a thread is still active.

Diagnostics with `trace`

While a workflow is executing, the orchestrator and its agents may emit trace events containing diagnostic and telemetry information, such as heartbeats, checkpoints, and execution progress. Use these events for debugging, logging, and displaying workflow progress in your application:

await thread.listenTrace { (frame: Any) in
    print("trace: \(frame)")
}

Complete example

func runWorkflow(adk: Adk) async {
    let orchestrator: Orchestrator = adk.orchestrator("trip-workflow")
    let thread: OrchestratorThread = await orchestrator.thread()

    await thread.listen { (data: [String: Any]) in
        guard let content = data["content"] as? [String: Any] else {
            return
        }

        // Respond to any question the workflow asks.
        if let reply = content["reply"] as? ([String: Any]) -> Void {
            reply([
                "user_input": "Budget 50k, 20-25 December"
            ])
            return
        }

        let type: String? = content["type"] as? String

        if type == "agent_general_response" {
            print("Done: \(content["message"] as? String ?? "")")
        } else if type == "agent_error_response" {
            print("Failed: \(content["message"] as? String ?? "")")
        }
    }

    await thread.listenTrace { (frame: Any) in
        print("working…")
    }

    await thread.push(
        event: "user_input",
        data: [
            "user_input": "Plan a 3-day trip to Goa"
        ]
    )
}

API reference

Member	Signature	Description
`threadId`	`String`	The thread identifier automatically attached to every outbound message.
`push`	`func push(event: String, data: [String: Any], options: PushConfig? = nil) async`	Sends an event to the workflow, automatically scoped to this thread.
`listen`	`func listen(_ callback: @escaping ([String: Any]) -> Void) async`	Registers a listener that receives every event as a raw `[event, content]` dictionary.
`bind`	`func bind(event: String, callback: @escaping (Any) -> Void)`	Registers a listener for a single event type. The callback receives the event's `content` directly.
`listenTrace`	`func listenTrace(_ callback: @escaping (Any) -> Void) async`	Registers a listener for `trace` diagnostic events.
`remove`	`func remove(event: String)`	Removes all listeners registered for the specified event.
`dispose`	`func dispose()`	Disposes the thread by unregistering it and detaching all associated listeners. Safe to call multiple times.
`isDisposed`	`Bool`	Whether the thread has been disposed.

Agent vs. Orchestrator

	Agent	Orchestrator
Best for	Communicating with a single AI agent	Executing multi-step workflows
Create a thread	`agent.thread()` (synchronous)	`await orchestrator.thread()`
Send input	`thread.run(input)` → `Run`	`thread.push(event, data)`
Receive results	`await run.done()` (single final response)	`thread.listen(...)` (event stream)
Event model	Strongly typed events	Raw event maps
Human-in-the-Loop	`run.sendFeedback(value)`	Injected `reply()` function
Lifecycle completion	The `Run` completes automatically	Call `thread.dispose()` when the workflow execution is complete

Prerequisites​

Workflow​

Quick start​

1. Connect to an orchestrator​

2. Start a thread​

3. Listen to the thread​

Listening to a single event​

4. Send the user's request​

Parameters​

5. Human-in-the-Loop (HITL)​

6. Clean up​

Diagnostics with trace​

Complete example​

API reference​

Agent vs. Orchestrator​