Orchestrator

The Orchestrator coordinates multiple agents to complete complex tasks. It analyzes the request, creates a plan, delegates subtasks to specialized agents, and collects results.

Orchestrator parameters

python

Orchestrator(
    model: Optional[str] = None,                 # Model name (e.g. "claude-sonnet-4-6")
    api_key: Optional[str] = None,               # API key (or set in .env)
    tools: Optional[list[str | Tool]] = None,    # Pre-built names or Tool instances
    agents: Optional[list[Agent]] = None,         # List of agents
    thinking: Optional[dict] = None,
    interaction_level: str = "balanced",         # "minimal", "balanced", "detailed"
    storage_dir: Optional[str] = None,

    # Clarification (off by default)
    clarification_prompt: Optional[str] = None,
    on_ask_user: Optional[Any] = None,

    # Delegation (off by default)
    allow_delegation: bool = False,
    delegation_timeout: int = 300,

    # Optional systems (all off by default)
    tracking: Optional[dict] = None,     # Cost tracking config
    recovery: Optional[dict] = None,     # Crash recovery config
    security: Optional[dict] = None,     # Sandbox config
    mcp_servers: Optional[list[dict]] = None,
    logging: Optional[dict] = None,
)

Basic usage

basic.py

from veska import Agent, Orchestrator

researcher = Agent(
    name="researcher",
    system_prompt="You research topics and provide detailed findings.",
    model="claude-sonnet-4-6",
)

writer = Agent(
    name="writer",
    system_prompt="You write clear, engaging content.",
    model="claude-sonnet-4-6",
)

orchestrator = Orchestrator(
    model="claude-sonnet-4-6",
    agents=[researcher, writer],
    tools=["file_manager"],
)

result = orchestrator.run("Write a research report on AI agents")

Workflow

When you call orchestrator.run(), it follows this flow:

1. ClarifyIf clarification_prompt is set, the orchestrator asks the user questions before planning.

2. PlanAI analyzes the request and creates a task plan with dependencies.

3. CheckpointIf checkpoints are configured, the plan is presented for approval.

4. ExecuteTasks are delegated to agents. Independent tasks run in parallel.

5. CollectResults are gathered and synthesized into a final response.

OrchestratorResult

python

OrchestratorResult(
    success: bool,
    plan: Optional[dict] = None,       # The execution plan
    results: Optional[dict] = None,    # Task + agent results
    progress: Optional[dict] = None,   # Final progress stats
    error: Optional[str] = None,
)

result.py

result = orchestrator.run("Build a todo app")

if result.success:
    print(result.plan)       # What the orchestrator planned
    print(result.results)    # Results from each task/agent
    print(result.progress)   # {"completed": 5, "total": 5, "percentage": 100}
else:
    print(f"Failed: {result.error}")

Clarification

The orchestrator can ask the user questions before planning:

clarification.py

from veska import Orchestrator

def ask_user(questions: str) -> str:
    print(questions)
    return input("Your answers: ")

orchestrator = Orchestrator(
    model="claude-sonnet-4-6",
    agents=[backend_agent, frontend_agent],
    clarification_prompt="""Before planning, ask the user about:
    - What database they want (PostgreSQL, MySQL, SQLite)
    - What frontend framework (React, Vue, plain HTML)
    - Any authentication requirements""",
    on_ask_user=ask_user,
)

Properties

Property	Description
`status`	idle, planning, running, paused, done, or failed
`progress`	Dict with completed, total, percentage
`message_bus`	MessageBus for agent communication
`events`	EventEmitter for app notifications
`shared_memory`	SharedMemory accessible by all agents
`task_planner`	TaskPlanner with dependency graph
`context`	ContextManager for history
`tool_registry`	ToolRegistry with all registered tools

Dynamic agent registration

# Register an agent after creation
orchestrator.register_agent(reviewer_agent)

# Get an agent by name
agent = orchestrator.get_agent("researcher")