functions.py

import asyncio
import os

import httpx

async def web_search(query: str, api_key: str):
    print(f"\n[Web Search] Searching for: {query}")

    if not api_key:
        return [{"error": "Brave API key is not configured"}]

    url = "https://api.search.brave.com/res/v1/web/search"
    headers = {
        "Accept": "application/json",
        "X-Subscription-Token": api_key,
    }
    params = {"q": query}

    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(url, headers=headers, params=params)
            response.raise_for_status()
            data = response.json()

        results = []
        for item in data.get("web", {}).get("results", [])[:5]:
            results.append(
                {
                    "title": item.get("title"),
                    "snippet": item.get("description"),
                    "url": item.get("url"),
                }
            )

        return results if results else [{"message": "No relevant results found"}]
    except Exception as e:
        return [{"error": f"Search request failed: {str(e)}"}]


async def execute_commands(type: str, command: list, blocked_commands: set):
    results = []
    print(f"\n[Tool] Running {len(command)} command(s) in {type}")

    for cmd in command:
        cmd_stripped = (cmd or "").strip()
        if not cmd_stripped:
            results.append({"cmd": cmd, "error": "Empty command"})
            continue

        first_word = cmd_stripped.split(maxsplit=1)[0].lower().rstrip(";")
        if first_word in blocked_commands:
            results.append(
                {
                    "cmd": cmd,
                    "error": f"Blocked potentially interactive command: {first_word}",
                }
            )
            print(f"  > Blocked: {cmd}")
            continue

        print(f"  > Running: {cmd}")
        try:
            if type == "bash":
                proc = await asyncio.create_subprocess_shell(
                    cmd,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
            elif type == "cmd":
                proc = await asyncio.create_subprocess_shell(
                    f"cmd.exe /c {cmd}",
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
            elif type == "pwsh":
                executable = "pwsh" if os.name != "nt" else "powershell"
                proc = await asyncio.create_subprocess_shell(
                    f"{executable} -Command {cmd}",
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                )
            else:
                results.append({"error": f"Unsupported shell type: {type}"})
                continue

            try:
                stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30.0)
            except asyncio.TimeoutError:
                proc.terminate()
                try:
                    await asyncio.wait_for(proc.wait(), timeout=5.0)
                except asyncio.TimeoutError:
                    proc.kill()
                results.append({"cmd": cmd, "error": "Command timed out after 30 seconds"})
                continue

            results.append(
                {
                    "cmd": cmd,
                    "stdout": stdout.decode("utf-8", errors="ignore").strip()[:1000],
                    "stderr": stderr.decode("utf-8", errors="ignore").strip()[:1000],
                    "returncode": proc.returncode,
                }
            )
        except Exception as e:
            results.append({"cmd": cmd, "error": str(e)})

    return results