dataflow.registry.discovery¶

Discovery mechanisms for the Haive Registry System.

This module provides functionality for automatically discovering and registering various components in the Haive ecosystem, such as agents, tools, engines, and games. It implements introspection mechanisms to find components based on naming conventions, class inheritance, and module structure.

The discovery process works by: 1. Searching for modules in specified paths 2. Inspecting classes in those modules 3. Determining if they match criteria for specific component types 4. Registering matching components in the registry system

Functions:

discover_modules: Find all Python modules under a base path discover_all: Discover all component types (agents, tools, engines, games) discover_agents: Discover and register agent components discover_tools: Discover and register tool components discover_toolkits: Discover and register toolkit components discover_engines: Discover and register engine components discover_games: Discover and register game components

Examples

Discovering components:

>>> from haive.dataflow.registry.discovery import discover_agents, discover_tools
>>>
>>> # Discover all agents in the system
>>> discovered_agents = discover_agents()
>>> print(f"Discovered {len(discovered_agents)} agents")
>>>
>>> # Discover tools and toolkits
>>> discovered_tools = discover_tools()
>>> discovered_toolkits = discover_toolkits()
>>> print(f"Discovered {len(discovered_tools)} tools and {len(discovered_toolkits)} toolkits")

Functions¶

discover_agents([module_paths])

Discover and register agents.

discover_all()

Discover and register all entity types.

discover_engines([module_paths])

Discover and register engines.

discover_games([module_paths])

Discover and register games.

discover_mcp_servers()

Discover and register MCP (Model Context Protocol) servers.

discover_modules(base_path)

Discover all Python modules under a base path.

discover_toolkits([module_paths])

Discover and register toolkits.

discover_tools([module_paths])

Discover and register tools.

is_pydantic_model(obj)

Check if an object is a Pydantic model.

Module Contents¶

dataflow.registry.discovery.discover_agents(module_paths=None)¶

Discover and register agents.

Parameters:

module_paths (list[str] | None) – Optional list of module paths to search

Returns:

List of registered agent IDs

Return type:

list[str]

dataflow.registry.discovery.discover_all()¶

Discover and register all entity types.

Returns:

Dictionary mapping entity types to lists of registered IDs

Return type:

dict[haive.dataflow.registry.models.EntityType, list[str]]

dataflow.registry.discovery.discover_engines(module_paths=None)¶

Discover and register engines.

Parameters:

module_paths (list[str] | None) – Optional list of module paths to search

Returns:

List of registered engine IDs

Return type:

list[str]

dataflow.registry.discovery.discover_games(module_paths=None)¶

Discover and register games.

Parameters:

module_paths (list[str] | None) – Optional list of module paths to search

Returns:

List of registered game IDs

Return type:

list[str]

dataflow.registry.discovery.discover_mcp_servers()¶

Discover and register MCP (Model Context Protocol) servers.

This function discovers MCP servers from various sources including: - npm packages (@modelcontextprotocol/) - PyPI packages (mcp-) - Local configurations - Existing haive-mcp downloaded servers

Returns:

List of registry IDs for registered MCP servers

Return type:

list[str]

Examples

>>> mcp_servers = discover_mcp_servers()
>>> print(f"Discovered {len(mcp_servers)} MCP servers")
dataflow.registry.discovery.discover_modules(base_path)¶

Discover all Python modules under a base path.

This function recursively explores a package to find all Python modules. It handles both regular modules and packages, traversing the entire module hierarchy to discover all available modules.

Parameters:

base_path (str) – Base module path to start discovery from (e.g., “haive.agents”)

Returns:

List of fully qualified module paths discovered

Return type:

List[str]

Examples

>>> modules = discover_modules("haive.tools")
>>> print(f"Discovered modules: {modules}")
Discovered modules: ['haive.tools.text', 'haive.tools.image', ...]
dataflow.registry.discovery.discover_toolkits(module_paths=None)¶

Discover and register toolkits.

Parameters:

module_paths (list[str] | None) – Optional list of module paths to search

Returns:

List of registered toolkit IDs

Return type:

list[str]

dataflow.registry.discovery.discover_tools(module_paths=None)¶

Discover and register tools.

Parameters:

module_paths (list[str] | None) – Optional list of module paths to search

Returns:

List of registered tool IDs

Return type:

list[str]

dataflow.registry.discovery.is_pydantic_model(obj)¶

Check if an object is a Pydantic model.

Parameters:

obj (Any) – Object to check

Returns:

True if it’s a Pydantic model, False otherwise

Return type:

bool