dataflow.api.routes.tools_routes_fixed¶

Tools API routes using the unified discovery system.

This module provides FastAPI routes for discovering and listing all available tools in the Haive ecosystem using the haive-core discovery system.

Attributes¶

`logger`
`router`

Classes¶

`ToolInfo`	Information about a tool.
`ToolInvokeRequest`	Request to invoke a tool.
`ToolInvokeResponse`	Response from tool invocation.
`ToolSchema`	Tool input/output schema information.
`ToolsListResponse`	Response for tools list endpoint.

Functions¶

`component_to_tool_info`(→ ToolInfo)	Convert a ComponentInfo to ToolInfo.
`discover_all_tools`(...)	Discover all tools using the unified discovery system.
`discover_tools_fallback`(...)	Fallback tool discovery if the main discovery fails.
`get_discovery_instance`(...)	Get or create the discovery instance.
`get_tool_categories`(→ dict[str, list[str]])	Get all available tool categories and their tools.
`get_tool_details`(→ dict[str, Any])	Get detailed information about a specific tool.
`get_tool_schema_endpoint`(→ ToolSchema)	Get the input/output schema for a specific tool.
`get_tool_stats`(→ dict[str, Any])	Get summary statistics about discovered tools.
`invoke_tool_endpoint`(→ ToolInvokeResponse)	Invoke a tool with the provided arguments.
`list_tools`(→ ToolsListResponse)	List all available tools.
`refresh_tool_cache`(→ dict[str, Any])	Refresh the tool discovery cache.
`search_tools`(→ ToolsListResponse)	Search for tools by query, category, or type.

Module Contents¶

class dataflow.api.routes.tools_routes_fixed.ToolInfo(/, **data: Any)¶

Bases: pydantic.BaseModel

Information about a tool.

category: str = None¶

description: str = None¶

metadata: dict[str, Any] = None¶

module: str = None¶

name: str = None¶

type: str = None¶

class dataflow.api.routes.tools_routes_fixed.ToolInvokeRequest(/, **data: Any)¶

Bases: pydantic.BaseModel

Request to invoke a tool.

arguments: dict[str, Any] = None¶

tool_name: str = None¶

class dataflow.api.routes.tools_routes_fixed.ToolInvokeResponse(/, **data: Any)¶

Bases: pydantic.BaseModel

Response from tool invocation.

error: str | None = None¶

result: Any = None¶

success: bool = None¶

class dataflow.api.routes.tools_routes_fixed.ToolSchema(/, **data: Any)¶

Bases: pydantic.BaseModel

Tool input/output schema information.

description: str = None¶

input_schema: dict[str, Any] = None¶

metadata: dict[str, Any] = None¶

name: str = None¶

output_schema: dict[str, Any] | None = None¶

class dataflow.api.routes.tools_routes_fixed.ToolsListResponse(/, **data: Any)¶

Bases: pydantic.BaseModel

Response for tools list endpoint.

count: int = None¶

discovery_method: str = None¶

tool_count: int = None¶

toolkit_count: int = None¶

tools: list[ToolInfo] = None¶

dataflow.api.routes.tools_routes_fixed.component_to_tool_info(component: haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo) → ToolInfo¶: Convert a ComponentInfo to ToolInfo.

dataflow.api.routes.tools_routes_fixed.discover_all_tools(force_refresh: bool = False) → list[haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo]¶: Discover all tools using the unified discovery system.

dataflow.api.routes.tools_routes_fixed.discover_tools_fallback() → list[haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo]¶: Fallback tool discovery if the main discovery fails.

dataflow.api.routes.tools_routes_fixed.get_discovery_instance() → haive.dataflow.api.routes.utils.haive_discovery.HaiveComponentDiscovery¶: Get or create the discovery instance.

async dataflow.api.routes.tools_routes_fixed.get_tool_categories() → dict[str, list[str]]¶

Get all available tool categories and their tools.

Returns:: Dictionary mapping categories to tool names

async dataflow.api.routes.tools_routes_fixed.get_tool_details(tool_name: str) → dict[str, Any]¶

Get detailed information about a specific tool.

Parameters:: tool_name – Name of the tool to get details for
Returns:: Detailed information about the tool

async dataflow.api.routes.tools_routes_fixed.get_tool_schema_endpoint(tool_name: str) → ToolSchema¶

Get the input/output schema for a specific tool.

Parameters:: tool_name – Name of the tool to get schema for
Returns:: ToolSchema containing input and output schemas

async dataflow.api.routes.tools_routes_fixed.get_tool_stats() → dict[str, Any]¶

Get summary statistics about discovered tools.

Returns:: Summary statistics

async dataflow.api.routes.tools_routes_fixed.invoke_tool_endpoint(request: ToolInvokeRequest) → ToolInvokeResponse¶

Invoke a tool with the provided arguments.

Parameters:: request – Tool invocation request with tool name and arguments
Returns:: ToolInvokeResponse with the result or error

async dataflow.api.routes.tools_routes_fixed.list_tools(force_refresh: bool = False) → ToolsListResponse¶

List all available tools.

Parameters:: force_refresh – Force refresh the tool cache
Returns:: ToolsListResponse containing list of available tools and total count

async dataflow.api.routes.tools_routes_fixed.refresh_tool_cache() → dict[str, Any]¶

Refresh the tool discovery cache.

Returns:: Status and count of discovered tools

async dataflow.api.routes.tools_routes_fixed.search_tools(query: str | None = None, category: str | None = None, tool_type: str | None = None) → ToolsListResponse¶

Search for tools by query, category, or type.

Parameters:

query – Search query to match against tool names and descriptions
category – Filter by category (e.g., ‘search’, ‘development’, ‘database’)
tool_type – Filter by type (‘tool’ or ‘toolkit’)

Returns:

ToolsListResponse containing filtered list of tools

dataflow.api.routes.tools_routes_fixed.logger¶

dataflow.api.routes.tools_routes_fixed.router¶