dataflow.api.routes.agent_discovery_routes_fixed

Agent discovery and management API routes using unified discovery system.

This module provides FastAPI routes for discovering and managing agents using the haive-core discovery system.

Attributes

logger
router

Classes

AgentCreateRequest	Request to create/instantiate an agent.
AgentCreateResponse	Response from agent creation.
AgentInfo	Information about an agent.
AgentListResponse	Response for agent list endpoint.
AgentSchema	Agent configuration/input schema information.

Functions

component_to_agent_info(→ AgentInfo)	Convert a ComponentInfo to AgentInfo.
discover_all_agents(...)	Discover all agents using the unified discovery system.
get_agent_details(→ dict[str, Any])	Get detailed information about a specific agent.
get_agent_schema(→ AgentSchema)	Get the configuration/initialization schema for a specific agent.
get_agent_stats(→ dict[str, Any])	Get summary statistics about discovered agents.
get_discovery_instance(...)	Get or create the discovery instance.
list_agents(→ AgentListResponse)	List all available agents.
refresh_agent_cache(→ dict[str, Any])	Refresh the agent discovery cache.
search_agents(→ AgentListResponse)	Search for agents by query, type, or category.

Module Contents

class dataflow.api.routes.agent_discovery_routes_fixed.AgentCreateRequest(/, **data: Any)

Bases: pydantic.BaseModel

Request to create/instantiate an agent.

agent_name: str = None

config: dict[str, Any] | None = None

init_args: dict[str, Any] | None = None

class dataflow.api.routes.agent_discovery_routes_fixed.AgentCreateResponse(/, **data: Any)

Bases: pydantic.BaseModel

Response from agent creation.

agent_id: str | None = None

agent_type: str | None = None

error: str | None = None

success: bool = None

class dataflow.api.routes.agent_discovery_routes_fixed.AgentInfo(/, **data: Any)

Bases: pydantic.BaseModel

Information about an agent.

agent_type: str = None

category: str = None

config_class: str | None = None

description: str = None

metadata: dict[str, Any] = None

module: str = None

name: str = None

version: str = None

class dataflow.api.routes.agent_discovery_routes_fixed.AgentListResponse(/, **data: Any)

Bases: pydantic.BaseModel

Response for agent list endpoint.

agents: list[AgentInfo] = None

count: int = None

discovery_method: str = None

v1_count: int = None

v2_count: int = None

class dataflow.api.routes.agent_discovery_routes_fixed.AgentSchema(/, **data: Any)

Bases: pydantic.BaseModel

Agent configuration/input schema information.

agent_type: str = None

config_schema: dict[str, Any] | None = None

description: str = None

init_schema: dict[str, Any] | None = None

metadata: dict[str, Any] = None

methods: list[str] = None

name: str = None

dataflow.api.routes.agent_discovery_routes_fixed.component_to_agent_info(component: haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo) → AgentInfo

Convert a ComponentInfo to AgentInfo.

dataflow.api.routes.agent_discovery_routes_fixed.discover_all_agents(force_refresh: bool = False) → list[haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo]

Discover all agents using the unified discovery system.

async dataflow.api.routes.agent_discovery_routes_fixed.get_agent_details(agent_name: str) → dict[str, Any]

Get detailed information about a specific agent.

Parameters:

agent_name – Name of the agent to get details for

Returns:

Detailed information about the agent

async dataflow.api.routes.agent_discovery_routes_fixed.get_agent_schema(agent_name: str) → AgentSchema

Get the configuration/initialization schema for a specific agent.

Parameters:

agent_name – Name of the agent to get schema for

Returns:

AgentSchema containing configuration and method information

async dataflow.api.routes.agent_discovery_routes_fixed.get_agent_stats() → dict[str, Any]

Get summary statistics about discovered agents.

Returns:

Summary statistics

dataflow.api.routes.agent_discovery_routes_fixed.get_discovery_instance() → haive.dataflow.api.routes.utils.haive_discovery.HaiveComponentDiscovery

Get or create the discovery instance.

async dataflow.api.routes.agent_discovery_routes_fixed.list_agents(force_refresh: bool = False) → AgentListResponse

List all available agents.

Parameters:

force_refresh – Force refresh the agent cache

Returns:

AgentListResponse containing list of available agents

async dataflow.api.routes.agent_discovery_routes_fixed.refresh_agent_cache() → dict[str, Any]

Refresh the agent discovery cache.

Returns:

Status and count of discovered agents

async dataflow.api.routes.agent_discovery_routes_fixed.search_agents(query: str | None = None, agent_type: str | None = None, category: str | None = None) → AgentListResponse

Search for agents by query, type, or category.

Parameters:

• query – Search query to match against agent names and descriptions

• agent_type – Filter by agent type (‘v1’ or ‘v2’)

• category – Filter by category

Returns:

AgentListResponse containing filtered list of agents

dataflow.api.routes.agent_discovery_routes_fixed.logger

dataflow.api.routes.agent_discovery_routes_fixed.router