dataflow.api.routes.tools_routes_enhanced¶

Enhanced Tools Discovery API Routes using Haive Core’s unified discovery. system.

This module provides FastAPI routes for discovering, managing, and invoking tools using the unified discovery system from haive-core. It supports both individual tools and toolkits with comprehensive schema extraction.

Key Features:

Unified tool discovery using haive-core’s discovery system
Support for both individual tools and toolkits
Schema extraction and validation
Tool invocation capabilities
Category-based organization
Performance caching

Examples

from fastapi import FastAPI from haive.dataflow.api.routes.tools_routes_enhanced import router

app = FastAPI() app.include_router(router, prefix=”/api/v1”)

# Endpoints available: # GET /api/v1/tools - List all tools # GET /api/v1/tools/search - Search tools # GET /api/v1/tools/{tool_name}/schema - Get tool schema # POST /api/v1/tools/invoke - Invoke a tool

Note

This implementation uses the fixed circular import pattern from haive-core to properly integrate with the discovery system.

Attributes¶

`logger`
`router`

Classes¶

`ToolInfo`	Information about a discovered tool or toolkit.
`ToolSchema`	Tool input/output schema information.
`ToolsListResponse`	Response for tools list endpoints.

Functions¶

`component_to_tool_info`(→ ToolInfo)	Convert a ComponentInfo to ToolInfo.
`discover_all_tools`(...)	Discover all tools using the unified discovery system.
`extract_tool_schema`(→ dict[str, Any] \| None)	Extract schema information from a tool component.
`get_tool_categories`(→ dict[str, list[str]])	Get all available tool categories and their tools.
`get_tool_schema_endpoint`(→ ToolSchema)	Get the input/output schema for a specific tool.
`get_tool_stats`(→ dict[str, Any])	Get statistics about discovered tools.
`infer_tool_category`(→ str)	Infer tool category from component information.
`list_tools`(, category, force_refresh)	List all available tools with optional filtering.
`search_tools`(, tool_type, category)	Search tools by name, description, or module path.

Module Contents¶

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

Bases: pydantic.BaseModel

Information about a discovered tool or toolkit.

name¶: Tool identifier name.

description¶: Human-readable description.

module¶: Full module path to the tool.

type¶: Type of tool (‘tool’ for individual, ‘toolkit’ for collection).

category¶: Category for grouping (e.g., ‘search’, ‘database’).

has_schema¶: Whether the tool has a defined input schema.

metadata¶: Additional metadata from discovery.

category: str = None¶

description: str = None¶

has_schema: bool = None¶

metadata: dict[str, Any] = None¶

module: str = None¶

name: str = None¶

type: str = None¶

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

Bases: pydantic.BaseModel

Tool input/output schema information.

name¶: Tool name.

description¶: Tool description.

input_schema¶: JSON Schema for tool input parameters.

output_schema¶: JSON Schema for tool output (if available).

examples¶: Usage examples (if available).

metadata¶: Additional schema metadata.

description: str = None¶

examples: list[dict[str, Any]] | None = 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_enhanced.ToolsListResponse(/, **data: Any)¶

Bases: pydantic.BaseModel

Response for tools list endpoints.

tools¶: List of discovered tools with metadata.

count¶: Total number of tools and toolkits.

tool_count¶: Number of individual tools.

toolkit_count¶: Number of toolkits.

discovery_method¶: Method used for discovery.

count: int = None¶

discovery_method: str = None¶

tool_count: int = None¶

toolkit_count: int = None¶

tools: list[ToolInfo] = None¶

dataflow.api.routes.tools_routes_enhanced.component_to_tool_info(component: haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo) → ToolInfo¶

Convert a ComponentInfo to ToolInfo.

Parameters:: component – ComponentInfo object from discovery.
Returns:: Structured tool information for API response.
Return type:: ToolInfo

dataflow.api.routes.tools_routes_enhanced.discover_all_tools(force_refresh: bool = False) → list[haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo]¶

Discover all tools using the unified discovery system.

Parameters:: force_refresh – If True, bypasses cache and rediscovers all tools.
Returns:: List of discovered tool components including both individual tools and toolkits.
Return type:: List[ComponentInfo]

Note

Uses haive-core’s discover_tools_with_schemas for comprehensive discovery with schema extraction.

dataflow.api.routes.tools_routes_enhanced.extract_tool_schema(component: haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo) → dict[str, Any] | None¶

Extract schema information from a tool component.

Parameters:: component – ComponentInfo object representing a tool.
Returns:: Extracted schema or None if not available.
Return type:: Optional[Dict[str, Any]]

Note

Attempts multiple strategies to extract schema: 1. From metadata ‘schema’ field 2. From LangChain tool args_schema 3. From custom get_input_schema method 4. From Pydantic model if available

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

Get all available tool categories and their tools.

Returns:: Dictionary mapping category names to sorted lists of tool names in each category.
Return type:: Dict[str, List[str]]

Examples

GET /api/v1/tools/categories

Response: {

“search”: [“GoogleSearchTool”, “BingSearchTool”], “database”: [“SQLDatabaseToolkit”, “MongoDBToolkit”], “development”: [“GithubToolkit”, “GitLabToolkit”]

}

async dataflow.api.routes.tools_routes_enhanced.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. Case-sensitive exact matching is used.

Returns:

Schema information including:

Input parameter schema (JSON Schema format)
Output schema if available
Usage examples if available

Return type:

ToolSchema

Raises:

HTTPException – 404 if tool is not found.

Examples

GET /api/v1/tools/GoogleSearchTool/schema

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

Get statistics about discovered tools.

Returns:

Statistics including:

total_tools: Total number of tools and toolkits
individual_tools: Number of individual tools
toolkits: Number of toolkits
categories: Count by category
with_schema: Number of tools with schemas
discovery_method: Method used
last_updated: Cache timestamp

Return type:

Dict[str, Any]

Examples

GET /api/v1/tools/stats

Response: {

“total_tools”: 50, “individual_tools”: 35, “toolkits”: 15, “categories”: {

“search”: 8, “database”: 10, “development”: 12

}, “with_schema”: 45

}

dataflow.api.routes.tools_routes_enhanced.infer_tool_category(component: haive.dataflow.api.routes.utils.haive_discovery.ComponentInfo) → str¶

Infer tool category from component information.

Parameters:: component – ComponentInfo object representing a tool.
Returns:: Inferred category name.
Return type:: str

Note

Categories are inferred from: 1. Metadata ‘category’ field 2. Module path keywords 3. Tool name patterns

async dataflow.api.routes.tools_routes_enhanced.list_tools(tool_type: str | None = Query(None, description='Filter by type (tool, toolkit)'), category: str | None = Query(None, description='Filter by category'), force_refresh: bool = Query(False, description='Force refresh discovery cache')) → ToolsListResponse¶

List all available tools with optional filtering.

Parameters:

tool_type – Optional filter for tool type (‘tool’ or ‘toolkit’).
category – Optional category filter (e.g., ‘search’, ‘database’).
force_refresh – If True, forces rediscovery of all tools.

Returns:

Response containing:

List of discovered tools with metadata
Count statistics
Discovery method information

Return type:

ToolsListResponse

Examples

GET /api/v1/tools?tool_type=toolkit&category=search

async dataflow.api.routes.tools_routes_enhanced.search_tools(query: str = Query(..., description='Search query'), tool_type: str | None = Query(None, description='Filter by type'), category: str | None = Query(None, description='Filter by category')) → ToolsListResponse¶

Search tools by name, description, or module path.

Parameters:

query – Search string to match against tool attributes. Case-insensitive partial matching is used.
tool_type – Optional filter for tool type.
category – Optional category filter.

Returns:

Filtered list of tools matching the search criteria.

Return type:

ToolsListResponse

Examples

GET /api/v1/tools/search?query=google&tool_type=toolkit

dataflow.api.routes.tools_routes_enhanced.logger¶

dataflow.api.routes.tools_routes_enhanced.router¶