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.

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(component)	Convert a ComponentInfo to AgentInfo.
discover_all_agents([force_refresh])	Discover all agents using the unified discovery system.
get_agent_details(agent_name)	Get detailed information about a specific agent.
get_agent_schema(agent_name)	Get the configuration/initialization schema for a specific agent.
get_agent_stats()	Get summary statistics about discovered agents.
get_discovery_instance()	Get or create the discovery instance.
list_agents([force_refresh])	List all available agents.
refresh_agent_cache()	Refresh the agent discovery cache.
search_agents([query, agent_type, category])	Search for agents by query, type, or category.

Module Contents

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

Bases: pydantic.BaseModel

Request to create/instantiate an agent.

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)

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

Bases: pydantic.BaseModel

Response from agent creation.

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)

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

Bases: pydantic.BaseModel

Information about an agent.

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)

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

Bases: pydantic.BaseModel

Response for agent list endpoint.

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)

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

Bases: pydantic.BaseModel

Agent configuration/input schema information.

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)

dataflow.api.routes.agent_discovery_routes_fixed.component_to_agent_info(component)

Convert a ComponentInfo to AgentInfo.

Parameters:

component (haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo)

Return type:

AgentInfo

dataflow.api.routes.agent_discovery_routes_fixed.discover_all_agents(force_refresh=False)

Discover all agents using the unified discovery system.

Parameters:

force_refresh (bool)

Return type:

list[haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo]

async dataflow.api.routes.agent_discovery_routes_fixed.get_agent_details(agent_name)

Get detailed information about a specific agent.

Parameters:

agent_name (str) – Name of the agent to get details for

Returns:

Detailed information about the agent

Return type:

dict[str, Any]

async dataflow.api.routes.agent_discovery_routes_fixed.get_agent_schema(agent_name)

Get the configuration/initialization schema for a specific agent.

Parameters:

agent_name (str) – Name of the agent to get schema for

Returns:

AgentSchema containing configuration and method information

Return type:

AgentSchema

async dataflow.api.routes.agent_discovery_routes_fixed.get_agent_stats()

Get summary statistics about discovered agents.

Returns:

Summary statistics

Return type:

dict[str, Any]

dataflow.api.routes.agent_discovery_routes_fixed.get_discovery_instance()

Get or create the discovery instance.

Return type:

haive.dataflow.api.routes.utils.haive_discovery.HaiveComponentDiscovery

async dataflow.api.routes.agent_discovery_routes_fixed.list_agents(force_refresh=False)

List all available agents.

Parameters:

force_refresh (bool) – Force refresh the agent cache

Returns:

AgentListResponse containing list of available agents

Return type:

AgentListResponse

async dataflow.api.routes.agent_discovery_routes_fixed.refresh_agent_cache()

Refresh the agent discovery cache.

Returns:

Status and count of discovered agents

Return type:

dict[str, Any]

async dataflow.api.routes.agent_discovery_routes_fixed.search_agents(query=None, agent_type=None, category=None)

Search for agents by query, type, or category.

Parameters:

• query (str | None) – Search query to match against agent names and descriptions

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

• category (str | None) – Filter by category

Returns:

AgentListResponse containing filtered list of agents

Return type:

AgentListResponse