agents.dynamic_supervisor.state

State schemas for dynamic supervisor agent.

from typing import Any This module defines the state management for the dynamic supervisor, including agent registry, routing control, and tool generation. Two versions are provided: - SupervisorState: Uses exclude=True for agent serialization (v1) - SupervisorStateV2: Attempts full agent serialization (experimental)

Classes:

SupervisorState: Base supervisor state with agent registry SupervisorStateWithTools: Extends base with dynamic tool generation SupervisorStateV2: Experimental version with full serialization

Examples

Creating and managing supervisor state:

state = SupervisorState()
state.add_agent("search", search_agent, "Search specialist")
state.activate_agent("search")

# List active agents
active = state.list_active_agents()

Classes

SupervisorState

Base state for dynamic supervisor with agent registry.

SupervisorStateV2

Experimental supervisor state with full agent serialization.

SupervisorStateWithTools

Supervisor state with dynamic tool generation.

Module Contents

class agents.dynamic_supervisor.state.SupervisorState(/, **data)

Bases: haive.core.schema.prebuilt.messages.messages_with_token_usage.MessagesStateWithTokenUsage

Base state for dynamic supervisor with agent registry.

Inherits from MessagesState for message handling and adds agent management capabilities. Agents are stored in a registry with metadata for routing.

Parameters:

data (Any)

agents

Registry of available agents by name

active_agents

List of currently active agent names (unique)

next_agent

Name of agent to execute next (routing control)

agent_task

Task to pass to the selected agent

agent_response

Response from the last executed agent

Examples

Basic agent management:

state = SupervisorState()
state.add_agent("math", math_agent, "Math specialist")
state.set_routing("math", "Calculate 2+2")

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

activate_agent(name)

Activate an inactive agent.

Parameters:

name (str) – Agent name to activate

Returns:

True if agent was activated, False if not found

Return type:

bool

add_agent(name, agent, description, active=True)

Add an agent to the registry.

Parameters:

  • name (str) – Unique identifier for the agent

  • agent (Any) – The agent instance

  • description (str) – Human-readable description of agent capabilities

  • active (bool) – Whether agent should be immediately active

Return type:

None

Examples

state.add_agent(“search”, search_agent, “Web search expert”, active=True)

clear_execution_state()

Clear execution state after completion.

Return type:

None

deactivate_agent(name)

Deactivate an active agent.

Parameters:

name (str) – Agent name to deactivate

Returns:

True if agent was deactivated, False if not found

Return type:

bool

classmethod ensure_unique_agents(v)

Ensure active agents list contains unique values.

Parameters:

v (list[str]) – List of agent names

Returns:

List with duplicates removed

Return type:

list[str]

get_agent(name)

Get agent instance by name.

Parameters:

name (str) – Agent name

Returns:

Agent instance or None if not found

Return type:

Any | None

list_active_agents()

List all active agents with descriptions.

Returns:

Dict mapping agent names to descriptions

Return type:

dict[str, str]

list_all_agents()

List all agents (active and inactive) with descriptions.

Returns:

Dict mapping agent names to descriptions

Return type:

dict[str, str]

remove_agent(name)

Remove an agent from the registry completely.

Parameters:

name (str) – Agent name to remove

Returns:

True if agent was removed, False if not found

Return type:

bool

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agents.dynamic_supervisor.state.SupervisorStateV2(/, **data)

Bases: haive.core.schema.prebuilt.messages.messages_with_token_usage.MessagesStateWithTokenUsage

Experimental supervisor state with full agent serialization.

This version attempts to serialize agents fully rather than excluding them. May require custom serialization logic or agent reconstruction.

Warning

This is experimental and may not work with all agent types or checkpointing systems. Use SupervisorState for production.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Parameters:

data (Any)

model_config

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class agents.dynamic_supervisor.state.SupervisorStateWithTools(/, **data)

Bases: SupervisorState

Supervisor state with dynamic tool generation.

Extends SupervisorState with automatic tool generation from registered agents. Creates handoff tools for each agent and manages a dynamic choice model.

Parameters:

data (Any)

agent_choice_model

Dynamic model for validated agent selection

generated_tools

List of tool names generated from agents

Examples

Using dynamic tools:

state = SupervisorStateWithTools()
state.add_agent("search", agent, "Search expert")
state.sync_agents()  # Generates handoff_to_search tool

tools = state.get_all_tools()  # Get tool instances

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

activate_agent(name)

Override to trigger tool regeneration.

Parameters:

name (str)

Return type:

bool

add_agent(name, agent, description, active=True)

Override to trigger tool regeneration.

Parameters:

  • name (str)

  • agent (Any)

  • description (str)

  • active (bool)

Return type:

None

deactivate_agent(name)

Override to trigger tool regeneration.

Parameters:

name (str)

Return type:

bool

get_all_tools()

Get all generated tools as callable instances.

Returns:

List of tool instances ready for use

Return type:

list[Any]

remove_agent(name)

Override to trigger tool regeneration.

Parameters:

name (str)

Return type:

bool

sync_agents()

Public method to sync agents with tools and choice model.

Call this after adding/removing agents to regenerate tools.

Return type:

None

sync_on_init()

Sync tools and choice model after initialization.

Return type:

SupervisorStateWithTools