tools.py

import asyncio
import subprocess
import textwrap

import httpx
from playwright.async_api import async_playwright

TIMEOUT_S = 15
MAX_OUTPUT_CHARS = 4000


async def run_python(code: str) -> str:
    """Executes arbitrary python in a subprocess with a timeout. Runs on the Space's
    own container, not the user's device -- fine for internal testing/automation,
    but do not expose this tool to untrusted callers."""
    script = textwrap.dedent(code)
    try:
        proc = await asyncio.create_subprocess_exec(
            "python3", "-c", script,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        try:
            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=TIMEOUT_S)
        except asyncio.TimeoutError:
            proc.kill()
            return "ERROR: execution timed out"
        out = stdout.decode(errors="replace") + stderr.decode(errors="replace")
        return out[:MAX_OUTPUT_CHARS] or "(no output)"
    except Exception as e:
        return f"ERROR: {e}"


async def test_endpoint(method: str, url: str, headers: dict | None = None, json_body: dict | None = None) -> str:
    """Direct HTTP call for testing your own API endpoints -- faster than spinning
    up a browser when you just need status/response body."""
    try:
        async with httpx.AsyncClient(timeout=TIMEOUT_S) as client:
            resp = await client.request(method.upper(), url, headers=headers, json=json_body)
            body = resp.text[:MAX_OUTPUT_CHARS]
            return f"status={resp.status_code}\nheaders={dict(resp.headers)}\nbody={body}"
    except Exception as e:
        return f"ERROR: {e}"


async def visit_url(url: str, extract: str = "text") -> str:
    """Loads a page with headless Chromium via Playwright. extract='text' returns
    visible page text, extract='html' returns raw HTML (truncated)."""
    try:
        async with async_playwright() as p:
            browser = await p.chromium.launch()
            page = await browser.new_page()
            await page.goto(url, timeout=TIMEOUT_S * 1000, wait_until="domcontentloaded")
            if extract == "html":
                content = await page.content()
            else:
                content = await page.inner_text("body")
            await browser.close()
            return content[:MAX_OUTPUT_CHARS]
    except Exception as e:
        return f"ERROR: {e}"