File size: 5,349 Bytes
e6853d3 359acf6 e6853d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | """ToolDiscoveryTool - Dynamic tool discovery for Stack 2.9"""
import json
from datetime import datetime
from typing import Any, Dict, List
from .base import BaseTool, ToolResult
from .registry import tool_registry
class ToolSearchTool(BaseTool):
"""Search for tools by capability or keyword."""
name = "tool_search"
description = "Search available tools by keyword or capability"
input_schema = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query (tool name, capability, or keyword)"
},
"capability": {
"type": "string",
"description": "Filter by capability (web, task, agent, data, etc.)"
}
},
"required": ["query"]
}
async def execute(self, query: str, capability: Optional[str] = None) -> ToolResult:
"""Search tools."""
all_tools = tool_registry.list_tools()
query_lower = query.lower()
matches = []
for name, tool_info in all_tools.items():
desc = tool_info.get("description", "").lower()
if query_lower in name.lower() or query_lower in desc:
matches.append({"name": name, **tool_info})
if capability:
cap_lower = capability.lower()
matches = [m for m in matches if cap_lower in m.get("description", "").lower()]
return ToolResult(success=True, data={
"query": query,
"capability": capability,
"matches": matches,
"count": len(matches)
})
class ToolListAllTool(BaseTool):
"""List all registered tools."""
name = "tool_list_all"
description = "List all available tools in the registry"
input_schema = {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "Filter by category (web, task, agent, team, skill, etc.)"
}
},
"required": []
}
async def execute(self, category: Optional[str] = None) -> ToolResult:
"""List all tools."""
all_tools = tool_registry.list_tools()
if category:
cat_lower = category.lower()
filtered = {k: v for k, v in all_tools.items() if cat_lower in v.get("description", "").lower()}
return ToolResult(success=True, data={
"category": category,
"tools": filtered,
"count": len(filtered)
})
return ToolResult(success=True, data={
"tools": all_tools,
"count": len(all_tools)
})
class ToolInfoTool(BaseTool):
"""Get detailed information about a specific tool."""
name = "tool_info"
description = "Get detailed information about a specific tool"
input_schema = {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Tool name"
}
},
"required": ["name"]
}
async def execute(self, name: str) -> ToolResult:
"""Get tool info."""
tool = tool_registry.get_tool(name)
if not tool:
return ToolResult(success=False, error=f"Tool '{name}' not found")
return ToolResult(success=True, data={
"name": tool.name,
"description": tool.description,
"input_schema": tool.input_schema
})
class ToolCapabilitiesTool(BaseTool):
"""List all available capabilities/functionalities."""
name = "tool_capabilities"
description = "List all tool capabilities grouped by category"
input_schema = {
"type": "object",
"properties": {},
"required": []
}
async def execute(self) -> ToolResult:
"""List capabilities."""
all_tools = tool_registry.list_tools()
categories = {
"web": ["web_search", "web_fetch", "web_fetch_meta"],
"task": ["task_create", "task_list", "task_update", "task_delete"],
"agent": ["agent_spawn", "agent_status", "agent_list"],
"team": ["team_create", "team_disband", "team_list", "team_status", "team_assign"],
"skill": ["skill_list", "skill_execute", "skill_info", "skill_chain", "skill_search"],
"scheduling": ["cron_create", "cron_list", "cron_delete"],
"messaging": ["message_send", "message_list", "message_channel", "message_template"],
"remote": ["remote_add", "remote_list", "remote_trigger", "remote_remove"],
"discovery": ["tool_search", "tool_list_all", "tool_info", "tool_capabilities"]
}
capabilities = {}
for cat, tools in categories.items():
available = [t for t in tools if t in all_tools]
if available:
capabilities[cat] = available
return ToolResult(success=True, data={
"capabilities": capabilities,
"total_categories": len(capabilities)
})
from typing import Optional
from .base import BaseTool
# Register tools
tool_registry.register(ToolSearchTool())
tool_registry.register(ToolListAllTool())
tool_registry.register(ToolInfoTool())
tool_registry.register(ToolCapabilitiesTool())
|