AI Toolbox

@dataclass
class Tool:
    id: str
    name: str
    description: str
    category: str
    parameters: list[ToolParameter]
    working_dir: str
    command_template: str
    output_type: str = "file"
    output_pattern: str = "*"

# Tool registry - declarative definitions
TOOLS: dict[str, Tool] = {
    "kokoro-tts": Tool(
        id="kokoro-tts",
        name="Kokoro TTS",
        category="Audio",
        working_dir="/projects/kokoro-tts",
        command_template=
            "docker run --rm -v {output_dir}:/out "
            "kokoro-tts:0.2 --voice {voice} --text '{text}' "
            "--out /out/{output_name}",
        parameters=[
            ToolParameter(name="text", type="textarea", ...),
            ToolParameter(name="voice", type="select", ...),
            ToolParameter(name="speed", type="number", ...),
        ],
    ),
    # ... more tools
}

async def run_job(job_id: str, tool: Tool) -> None:
    """Execute a tool job asynchronously."""
    job = get_job(job_id)
    job["status"] = "running"
    job["started_at"] = datetime.now().isoformat()
    
    # Build command from template
    cmd, work_dir = build_command(tool, job["parameters"], job_id)
    
    # Run in subprocess with asyncio
    process = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=work_dir,
    )
    
    stdout, stderr = await process.communicate()
    
    if process.returncode == 0:
        job["status"] = "completed"
        job["output"] = {"files": find_output_files(job_id)}
    else:
        job["status"] = "failed"
        job["error"] = stderr.decode()