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.
Orchestrator orchestrator = adk.orchestrator('your-orchestrator-id');

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

// 3. Listen for the events the workflow sends back.
thread.listen((data) {
  final content = data['content'];
  if (content is Map && content['type'] == 'agent_general_response') {
    print('Answer: ${content['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():

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:

OrchestratorThread thread = 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:

OrchestratorThread resumed = await orchestrator.thread(threadId: 'thread_1753070000000_1a2b3c');

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']:

thread.listen((Map<String, dynamic> data) {
  final content = data['content'];
  final type = content is Map ? content['type'] : null;

  switch (type) {
    case 'agent_general_response':
      print('Answer: ${content['message']}');
      break;
    case 'human_input_request':
      print('Workflow asks: ${content['prompt']}');
      break;
    case 'agent_error_response':
      print('Error: ${content['message']}');
      break;
    case 'agent_wait_response':
      print('Waiting on: ${content['waiting_for_agent_id']}');
      break;
    case 'planner_correction_request':
      print('Revising plan: ${content['reason']}');
      break;
    default:
      print('Event: ${data['event']}');
      break;
  }
});

The callback argument

Each invocation of the listener receives a raw event map with the following fields:

Field	Type	Description
`data['event']`	`String`	The wire-level event name.
`data['content']`	`dynamic`	The event payload. This is typically a `Map<String, dynamic>` whose fields you access directly (for example, `content['message']` or `content['prompt']`).

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',
  callback: (content) => print('status: $content'),
);

thread.remove(event: 'status_update'); // stop listening to this event

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 a user_input event containing your application data. The OrchestratorThread automatically attaches the thread_id to every outbound message:

await thread.push(
  event: 'user_input',
  data: {'user_input': 'Plan a 3-day trip to Goa'},
);

The contents of the data map are defined by your workflow. Include whatever fields your workflow expects:

await thread.push(
  event: 'user_input',
  data: {
    'user_input': 'Plan a trip',
    'preferences': {'budget': 50000, 'days': 3},
  },
);

Parameters

Parameter	Type	Description
`event`	`String`	The event name, for example `user_input` or `user_reply`.
`data`	`Map<String, dynamic>`	The event payload. Its structure is defined by your workflow.
`options`	`PushConfig?`	Optional delivery configuration. The `OrchestratorThread` automatically sets the `thread_id` 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) {
  final content = data['content'];

  if (content is Map && content['reply'] is Function) {
    print('Workflow asks: ${content['prompt']}');

    // Collect the answer from your UI, then call reply.
    final reply = content['reply'] as Function;
    reply({'user_input': 'Budget 50k, December 20-25'});
    return;
  }

  // …handle other events…
});

The reply function is injected only into human-input requests — that is, when the event is human_input_request or content['type'] is human_input_request.

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:

thread.listenTrace((dynamic frame) {
  print('trace: $frame');
});

Complete example

Future<void> runWorkflow(Adk adk) async {
  Orchestrator orchestrator = adk.orchestrator('trip-workflow');
  OrchestratorThread thread = await orchestrator.thread();

  thread.listen((Map<String, dynamic> data) {
    final content = data['content'];

    // Respond to any question the workflow asks.
    if (content is Map && content['reply'] is Function) {
      (content['reply'] as Function)({'user_input': 'Budget 50k, 20-25 December'});
      return;
    }

    final type = content is Map ? content['type'] : null;
    if (type == 'agent_general_response') {
      print('Done: ${content['message']}');
    } else if (type == 'agent_error_response') {
      print('Failed: ${content['message']}');
    }
  });

  thread.listenTrace((dynamic frame) => 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`	`Future<void> push({required String event, required Map<String, dynamic> data, PushConfig? options})`	Sends an event to the workflow, automatically scoped to this thread.
`listen`	`void listen(void Function(Map<String, dynamic> data) callback)`	Registers a listener that receives every event as a raw `{event, content}` map.
`bind`	`void bind({required String event, required void Function(dynamic data) callback})`	Registers a listener for a single event type. The callback receives the event's `content` directly.
`listenTrace`	`void listenTrace(void Function(dynamic data) callback)`	Registers a listener for `trace` diagnostic events.
`remove`	`void remove({required String event})`	Removes all listeners registered for the specified event.
`dispose`	`void 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​

The callback argument​

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​