42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
from typing import Any
|
|
|
|
from duck_core.tools.base import Tool, ToolResult
|
|
from duck_core.tools.file_read import FileReadTool
|
|
from duck_core.tools.file_write import FileWriteTool
|
|
from duck_core.tools.list_dir import ListDirTool
|
|
from duck_core.tools.search_files import SearchFilesTool
|
|
from duck_core.tools.shell_exec_safe import ShellExecSafeTool
|
|
|
|
|
|
class ToolGateway:
|
|
def __init__(self, tools: list[Tool]):
|
|
self.tools = {tool.name: tool for tool in tools}
|
|
|
|
@classmethod
|
|
def default(cls, workspace: str) -> "ToolGateway":
|
|
return cls(
|
|
[
|
|
FileReadTool(workspace),
|
|
FileWriteTool(workspace),
|
|
ListDirTool(workspace),
|
|
SearchFilesTool(workspace),
|
|
ShellExecSafeTool(workspace),
|
|
]
|
|
)
|
|
|
|
async def run_action(
|
|
self, action: dict[str, Any], approved: bool = False, password: str | None = None
|
|
) -> ToolResult:
|
|
tool_name = str(action.get("tool", ""))
|
|
tool = self.tools.get(tool_name)
|
|
if tool is None:
|
|
return ToolResult(ok=False, error=f"Unknown tool: {tool_name}")
|
|
args = action.get("args") or {}
|
|
if not isinstance(args, dict):
|
|
return ToolResult(ok=False, error="Tool args must be an object")
|
|
if approved:
|
|
args = {**args, "_approved": True}
|
|
if password is not None:
|
|
args = {**args, "_password": password}
|
|
return await tool.run(args)
|