A3S Docs
A3S Code

Agent Orchestrator

Main-Sub Agent Coordinator - Real-time monitoring and dynamic control of multiple SubAgents

Agent Orchestrator

Agent Orchestrator is a main-sub agent coordinator built on a3s-event, providing a unified event bus for real-time monitoring of all sub-agent behaviors, planning, and execution, with dynamic control capabilities.

Version Requirement: v1.1.0+

Core Features

  • Real AgentLoop Execution - SubAgents run real LLM + tool loops when Orchestrator.create(agent) is used
  • Real-time Monitoring - Monitor all SubAgent lifecycles, state changes, progress, and tool executions
  • Task List Query - View all SubAgent task lists and current activities in real-time
  • Activity Tracking - Track each SubAgent's current activity (calling tools, requesting LLM, waiting for control)
  • Dynamic Control - Pause, resume, cancel SubAgents
  • Event-Driven - Unified event bus based on a3s-event
  • Pluggable Communication - Default memory-based events, supports custom NATS provider
  • Concurrency Management - Control maximum concurrent SubAgent count

Architecture

AgentOrchestrator (Main Agent)
    ↓ Subscribe Events
┌───────────────────────────────┐
│   a3s-event EventBus          │
│ (Memory / NATS / Custom)      │
└───────────────────────────────┘
    ↑ Publish Events    ↓ Control Signals
SubAgentWrapper (Sub Agents)

Quick Start

import { Orchestrator, SubAgentConfig } from '@a3s-lab/code';

// Create orchestrator
const orch = Orchestrator.create();

// Create SubAgent config
const config: SubAgentConfig = {
  agentType: 'general',
  prompt: 'Use glob to find TypeScript files',
  description: 'Find TypeScript files',
  permissive: true,
  maxSteps: 10,
};

// Spawn SubAgent
const handle = orch.spawnSubagent(config);
console.log(`SubAgent ID: ${handle.id}`);

// Monitor state
for (let i = 0; i < 10; i++) {
  await new Promise(resolve => setTimeout(resolve, 500));
  const state = handle.state();
  const active = orch.activeCount();
  console.log(`[${i * 0.5}s] Active: ${active}, State: ${state}`);

  // Control operations
  if (i === 3) {
    handle.pause();
    console.log('Paused');
  }
  if (i === 5) {
    handle.resume();
    console.log('Resumed');
  }
}

// Wait for completion
const result = await handle.wait();
console.log(`Result: ${result}`);
from a3s_code import Orchestrator, SubAgentConfig
import time

# Create orchestrator
orch = Orchestrator.create()

# Create SubAgent config
config = SubAgentConfig(
    agent_type="general",
    prompt="Use glob to find Python files",
    description="Find Python files",
    permissive=True,
    max_steps=10
)

# Spawn SubAgent
handle = orch.spawn_subagent(config)
print(f"SubAgent ID: {handle.id}")

# Monitor state
for i in range(10):
    time.sleep(0.5)
    state = handle.state()
    active = orch.active_count()
    print(f"[{i*0.5}s] Active: {active}, State: {state}")

    # Control operations
    if i == 3:
        handle.pause()
        print("Paused")
    if i == 5:
        handle.resume()
        print("Resumed")

# Wait for completion
result = handle.wait()
print(f"Result: {result}")

Real Execution

Orchestrator.create() without an agent spawns subagents using an internal stub — for tests and demos only. To run real LLM + tool execution, pass a real Agent instance to Orchestrator.create(agent).

import { Agent, Orchestrator } from '@a3s-lab/code';

const agent = await Agent.create('agent.hcl');

// Wire the orchestrator to real AgentLoop execution
const orch = Orchestrator.create(agent);

const handle = orch.spawnSubagent({
  agentType: 'general',
  prompt: 'Use glob to find all TypeScript files. Report how many you found.',
  permissive: true,
  maxSteps: 5,
});

// Block until done
orch.waitAll();
from a3s_code import Agent, Orchestrator, SubAgentConfig

# 1. Load your agent config (any provider — Kimi, Claude, OpenAI, …)
agent = Agent.create("agent.hcl")

# 2. Wire the orchestrator to real AgentLoop execution
orch = Orchestrator.create(agent)

# 3. Spawn and run — uses real LLM + tool calls
handle = orch.spawn_subagent(SubAgentConfig(
    agent_type="general",
    prompt="Use glob to find all Python files. Report how many you found.",
    permissive=True,
    max_steps=5,
))

result = handle.wait()
print(result)

Execution Mode Comparison

ModeAPISubagent runs
StubOrchestrator.create()Internal placeholder — for tests & demos
Real executionOrchestrator.create(agent)Full AgentLoop with real LLM + tools

Architecture with Real Execution

AgentOrchestrator::from_agent(agent)
    └── SubAgentWrapper  (Agent attached)
            ├── AgentRegistry   — looks up agent type definition
            ├── ToolExecutor    — built-in tools only (no Task tool to prevent nesting)
            └── AgentLoop       — real LLM turns + tool execution
                    │ events

        OrchestratorEvent::SubAgentInternalEvent { id, event }
  • Timeout: Set timeout_ms to abort long-running subagents; the AgentLoop task is cancelled and an error returned
  • Pause: Marks Paused and stops accepting control signals; the current LLM turn finishes before stopping. cancel() aborts immediately
  • duration_ms: Real elapsed time from spawn to completion, not hardcoded zero

Event Types

AgentEvent is a flat interface with a type string field. Subscribe via handle.events() which returns a SubAgentEventStream.

Event type (event.type)FieldsDescription
startprompt?SubAgent began execution
text_deltatextText chunk from assistant
turn_startturnNew turn started
turn_endturn, totalTokens?Turn completed
tool_starttoolName, toolIdTool call started
tool_endtoolName, toolId, toolOutput?, exitCode?Tool call finished
endtotalTokens?Generation finished
errorerrorError occurred
btw_answerquestion, answerEphemeral side question answered

SubAgent States

Control Signals

MethodDescription
handle.pause()Pause execution; can resume later
handle.resume()Resume a paused SubAgent
handle.cancel()Abort immediately, cannot resume
handle.wait()Block until SubAgent completes and return result
handle.events()Subscribe to SubAgent event stream

adjust_params and inject_prompt shown in older documentation do not exist in the current SDK.

Real-time Monitoring API

// Get all SubAgent information list
const subagents = orch.listSubagents();
for (const info of subagents) {
  console.log(`SubAgent: ${info.id}`);
  console.log(`  Type: ${info.agentType}`);
  console.log(`  State: ${info.state}`);
  console.log(`  Description: ${info.description}`);

  if (info.currentActivity) {
    console.log(`  Current Activity: ${info.currentActivity.activityType}`);
  }
}

// Get specific SubAgent details
const info = orch.getSubagentInfo('subagent-1');
if (info) {
  console.log(`Created at: ${info.createdAt}`);
  console.log(`Updated at: ${info.updatedAt}`);
}

// Get all active SubAgent activities
const activities = orch.getActiveActivities();
for (const entry of activities) {
  if (entry.activity.activityType === 'calling_tool') {
    console.log(`${entry.id} is calling a tool`);
  } else if (entry.activity.activityType === 'requesting_llm') {
    console.log(`${entry.id} is requesting LLM`);
  }
}

// Get all SubAgent states
const states = orch.getAllStates();
for (const entry of states) {
  console.log(`${entry.id}: ${entry.state}`);
}

// Get active count
const count = orch.activeCount();
console.log(`Active SubAgents: ${count}`);
# Get all SubAgent information list
subagents = orch.list_subagents()
for info in subagents:
    print(f"SubAgent: {info.id}")
    print(f"  Type: {info.agent_type}")
    print(f"  State: {info.state}")
    print(f"  Description: {info.description}")

    if info.current_activity:
        print(f"  Current Activity: {info.current_activity.activity_type}")

# Get specific SubAgent details
info = orch.get_subagent_info("subagent-1")
if info:
    print(f"Created at: {info.created_at}")
    print(f"Updated at: {info.updated_at}")

# Get all active SubAgent activities
activities = orch.get_active_activities()
for subagent_id, activity in activities:
    if activity.activity_type == "calling_tool":
        print(f"{subagent_id} is calling a tool")
    elif activity.activity_type == "requesting_llm":
        print(f"{subagent_id} is requesting LLM")

# Get all SubAgent states
states = orch.get_all_states()
for subagent_id, state in states:
    print(f"{subagent_id}: {state}")

# Get active count
count = orch.active_count()
print(f"Active SubAgents: {count}")

SubAgent Activity Types

SubAgents have the following activity types during execution:

Activity TypeDescriptionData
IdleIdle state-
CallingToolCalling a toolTool name and arguments
RequestingLlmRequesting LLMMessage count
WaitingForControlWaiting for control signalReason

Monitoring API List

APIDescription
list_subagents()Get all SubAgent information list
get_subagent_info(id)Get specific SubAgent details
get_active_activities()Get all active SubAgent activities
get_all_states()Get all SubAgent states
active_count()Get active SubAgent count
pause_subagent(id)Pause specified SubAgent
resume_subagent(id)Resume specified SubAgent
cancel_subagent(id)Cancel specified SubAgent
wait_all()Wait for all SubAgents to complete

Configuration Options

OrchestratorConfig

SubAgentConfig

Advanced Usage

Subscribe to Specific SubAgent Events

Concurrency Limits

Nested SubAgents

Comparison with AgentTeam

For team-based multi-agent collaboration (Lead/Worker/Reviewer workflow), see Agent Teams.

FeatureAgentTeamAgentOrchestrator
ArchitecturePeer collaborationMain-sub coordination
Communicationmpsc + shared TaskBoardEvent bus (broadcast)
ControlAutonomousCentralized
MonitoringTask-levelReal-time event stream
Use CaseMulti-role collaborationMain agent supervision

Best Practices

  1. Use create_with_agent for production - Always wire a real agent for actual LLM execution; create() alone only runs stubs
  2. Set Reasonable Concurrency Limits - Configure max_concurrent_subagents based on system resources
  3. Use Event Filtering - Subscribe to specific SubAgent events to reduce noise
  4. Handle Error States - Listen to SubAgentStateChanged events and handle errors
  5. Set Timeouts - Configure timeout_ms for long-running tasks; the underlying task is aborted automatically on timeout
  6. Clean Up Resources - Use cancel() to clean up unneeded SubAgents

Examples

For complete examples, see:

On this page