dataflow.api.registry

Attributes

agent_registry
logger

Classes

AgentRegistryService

Service that manages agent registration and discovery.

Module Contents

class dataflow.api.registry.AgentRegistryService

Service that manages agent registration and discovery.

discover_agents(search_paths: list[str] | None = None) → None

Automatically discover and register all agent configurations from. multiple paths.

Parameters:

search_paths (list[str] | None) – List of package paths to search for agents

Return type:

None

get_agent_config(name: str) → type[haive.core.engine.aug_llm.AugLLMConfig] | None

Get agent configuration class by name.

Parameters:

name (str)

Return type:

type[haive.core.engine.aug_llm.AugLLMConfig] | None

get_agent_type(name: str) → str | None

Get agent type by name.

Parameters:

name (str)

Return type:

str | None

get_or_create_agent(name: str, thread_id: str | None = None, **config_kwargs) → tuple[haive.agents.base.agent.Agent | None, str | None]

Get a previously instantiated agent or create a new one. Returns the. agent and an error message if there was a problem.

Parameters:

• name (str) – Agent name

• thread_id (str | None) – Optional thread ID for persistence

• **config_kwargs – Configuration parameters for agent creation

Returns:

Tuple of (agent, error_message)

Return type:

tuple[haive.agents.base.agent.Agent | None, str | None]

list_available_agents() → list[str]

List all available agent configurations.

Return type:

list[str]

list_failed_agents() → dict[str, str]

List agents that failed to register with their error messages.

Return type:

dict[str, str]

register_agent_config(name: str, config_class: type[haive.core.engine.aug_llm.AugLLMConfig], agent_type: str = 'agent') → None

Register an agent configuration class.

Parameters:

• name (str) – Name of the agent

• config_class (type[haive.core.engine.aug_llm.AugLLMConfig]) – Agent configuration class

• agent_type (str) – Type of agent (e.g., ‘agent’, ‘game’)

Return type:

None

agent_configs: dict[str, dict[str, Any]]

agent_errors: dict[str, str]

db_connection = None

default_persistence_type = 'postgres'

instantiated_agents: dict[str, haive.agents.base.agent.Agent]

dataflow.api.registry.agent_registry

dataflow.api.registry.logger