Serve

serve

HTTP server mode for wavexis using aiohttp.

aiohttp is an optional dependency under the [serve] extra. All imports are lazy — WavexisError is raised if aiohttp is not installed.

TokenBucket

Token bucket rate limiter for the HTTP API.

Allows up to capacity requests per refill_period seconds. Tokens refill continuously at a rate of capacity/refill_period per second.

Source code in wavexis/serve.py

class TokenBucket:
    """Token bucket rate limiter for the HTTP API.

    Allows up to `capacity` requests per `refill_period` seconds.
    Tokens refill continuously at a rate of capacity/refill_period per second.
    """

    def __init__(self, capacity: int, refill_period: float) -> None:
        """Initialize the token bucket.

        Args:
            capacity: Maximum number of tokens (burst size).
            refill_period: Seconds to fully refill from empty.

        Raises:
            ValueError: If capacity or refill_period are not valid.
        """
        if not isinstance(capacity, int) or capacity < 1:
            raise ValueError("capacity must be a positive integer")
        if refill_period <= 0:
            raise ValueError("refill_period must be positive")
        self._capacity = capacity
        self._tokens = float(capacity)
        self._refill_rate = capacity / refill_period
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self) -> bool:
        """Try to acquire a token.

        Returns:
            True if a token was acquired, False if rate limited.
        """
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(self._capacity, self._tokens + elapsed * self._refill_rate)
            self._last_refill = now
            if self._tokens >= 1:
                self._tokens -= 1
                return True
            return False

    async def retry_after(self) -> float:
        """Return seconds until the next token is available."""
        async with self._lock:
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(self._capacity, self._tokens + elapsed * self._refill_rate)
            self._last_refill = now
            if self._tokens >= 1:
                return 0.0
            if self._refill_rate <= 0:
                return 1.0
            return (1 - self._tokens) / self._refill_rate

__init__

__init__(capacity: int, refill_period: float) -> None

Initialize the token bucket.

Parameters:

Name	Type	Description	Default
capacity	int	Maximum number of tokens (burst size).	required
refill_period	float	Seconds to fully refill from empty.	required

Raises:

Type	Description
ValueError	If capacity or refill_period are not valid.

Source code in wavexis/serve.py

def __init__(self, capacity: int, refill_period: float) -> None:
    """Initialize the token bucket.

    Args:
        capacity: Maximum number of tokens (burst size).
        refill_period: Seconds to fully refill from empty.

    Raises:
        ValueError: If capacity or refill_period are not valid.
    """
    if not isinstance(capacity, int) or capacity < 1:
        raise ValueError("capacity must be a positive integer")
    if refill_period <= 0:
        raise ValueError("refill_period must be positive")
    self._capacity = capacity
    self._tokens = float(capacity)
    self._refill_rate = capacity / refill_period
    self._last_refill = time.monotonic()
    self._lock = asyncio.Lock()

acquire async

acquire() -> bool

Try to acquire a token.

Returns:

Type	Description
bool	True if a token was acquired, False if rate limited.

Source code in wavexis/serve.py

async def acquire(self) -> bool:
    """Try to acquire a token.

    Returns:
        True if a token was acquired, False if rate limited.
    """
    async with self._lock:
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(self._capacity, self._tokens + elapsed * self._refill_rate)
        self._last_refill = now
        if self._tokens >= 1:
            self._tokens -= 1
            return True
        return False

retry_after async

retry_after() -> float

Return seconds until the next token is available.

Source code in wavexis/serve.py

async def retry_after(self) -> float:
    """Return seconds until the next token is available."""
    async with self._lock:
        now = time.monotonic()
        elapsed = now - self._last_refill
        self._tokens = min(self._capacity, self._tokens + elapsed * self._refill_rate)
        self._last_refill = now
        if self._tokens >= 1:
            return 0.0
        if self._refill_rate <= 0:
            return 1.0
        return (1 - self._tokens) / self._refill_rate

BackendPool

Concurrency limiter and connection pool for browser backends.

Uses a semaphore to cap the number of simultaneous browser instances. Maintains a pool of reusable backend instances to avoid launching a new browser per request.

get_backend acquires a slot and return_backend/discard_backend release it, so callers cannot leak the semaphore if backend creation fails.

Source code in wavexis/serve.py

class BackendPool:
    """Concurrency limiter and connection pool for browser backends.

    Uses a semaphore to cap the number of simultaneous browser instances.
    Maintains a pool of reusable backend instances to avoid launching a
    new browser per request.

    ``get_backend`` acquires a slot and ``return_backend``/``discard_backend``
    release it, so callers cannot leak the semaphore if backend creation fails.
    """

    def __init__(self, max_concurrent: int = 5) -> None:
        self._max_concurrent = max_concurrent
        self._semaphore = asyncio.Semaphore(max_concurrent)
        self._pool: asyncio.Queue[AbstractBackend] = asyncio.Queue(maxsize=max_concurrent)
        self._created: int = 0
        self._lock = asyncio.Lock()

    async def get_backend(
        self,
        preferred: str | None = None,
    ) -> AbstractBackend:
        """Acquire a slot and get a backend from the pool or create a new one.

        Reuses an idle backend if available, otherwise creates a new one.
        The acquired slot is released by ``return_backend`` or
        ``discard_backend``.

        Args:
            preferred: Preferred backend name for new instances.

        Returns:
            A backend instance (may or may not be launched yet).
        """
        await self._semaphore.acquire()
        async with self._lock:
            if not self._pool.empty():
                return self._pool.get_nowait()
            self._created += 1
        try:
            return await get_manager().select_with_fallback(preferred)
        except Exception:
            # Creation failed — release the slot and accounting so the pool
            # doesn't slowly fill with phantom backends.
            async with self._lock:
                self._created -= 1
            self._semaphore.release()
            raise

    async def return_backend(self, backend: AbstractBackend) -> None:
        """Return a backend to the pool for reuse and release its slot.

        The backend is returned without closing it so it can be reused
        by subsequent requests. Backends are closed only by ``close_all``
        during shutdown or when the pool is full.

        Args:
            backend: The backend instance to return.
        """
        # If the pool is full, close the backend instead of queueing it.
        if self._pool.full():
            with contextlib.suppress(Exception):
                await backend.close()
            async with self._lock:
                self._created -= 1
            self._semaphore.release()
            return
        await _sanitize_backend(backend)
        await self._pool.put(backend)
        self._semaphore.release()

    async def discard_backend(self, backend: AbstractBackend) -> None:
        """Close a broken backend and release its slot.

        Use this when a backend failed to launch or is in an unknown state
        and must not be reused.

        Args:
            backend: The backend instance to discard.
        """
        with contextlib.suppress(Exception):
            await backend.close()
        async with self._lock:
            self._created -= 1
        self._semaphore.release()

    async def close_all(self) -> None:
        """Close all pooled backends, drain the pool, and reset all slots."""
        while not self._pool.empty():
            backend = self._pool.get_nowait()
            with contextlib.suppress(Exception):
                await backend.close()
        async with self._lock:
            self._created = 0
        # Reset the semaphore so leftover acquired slots do not leak between
        # lifecycles (e.g., across tests).
        self._semaphore = asyncio.Semaphore(self._max_concurrent)

get_backend async

get_backend(preferred: str | None = None) -> AbstractBackend

Acquire a slot and get a backend from the pool or create a new one.

Reuses an idle backend if available, otherwise creates a new one. The acquired slot is released by return_backend or discard_backend.

Parameters:

Name	Type	Description	Default
preferred	str | None	Preferred backend name for new instances.	None

Returns:

Type	Description
AbstractBackend	A backend instance (may or may not be launched yet).

Source code in wavexis/serve.py

async def get_backend(
    self,
    preferred: str | None = None,
) -> AbstractBackend:
    """Acquire a slot and get a backend from the pool or create a new one.

    Reuses an idle backend if available, otherwise creates a new one.
    The acquired slot is released by ``return_backend`` or
    ``discard_backend``.

    Args:
        preferred: Preferred backend name for new instances.

    Returns:
        A backend instance (may or may not be launched yet).
    """
    await self._semaphore.acquire()
    async with self._lock:
        if not self._pool.empty():
            return self._pool.get_nowait()
        self._created += 1
    try:
        return await get_manager().select_with_fallback(preferred)
    except Exception:
        # Creation failed — release the slot and accounting so the pool
        # doesn't slowly fill with phantom backends.
        async with self._lock:
            self._created -= 1
        self._semaphore.release()
        raise

return_backend async

return_backend(backend: AbstractBackend) -> None

Return a backend to the pool for reuse and release its slot.

The backend is returned without closing it so it can be reused by subsequent requests. Backends are closed only by close_all during shutdown or when the pool is full.

Parameters:

Name	Type	Description	Default
backend	AbstractBackend	The backend instance to return.	required

Source code in wavexis/serve.py

async def return_backend(self, backend: AbstractBackend) -> None:
    """Return a backend to the pool for reuse and release its slot.

    The backend is returned without closing it so it can be reused
    by subsequent requests. Backends are closed only by ``close_all``
    during shutdown or when the pool is full.

    Args:
        backend: The backend instance to return.
    """
    # If the pool is full, close the backend instead of queueing it.
    if self._pool.full():
        with contextlib.suppress(Exception):
            await backend.close()
        async with self._lock:
            self._created -= 1
        self._semaphore.release()
        return
    await _sanitize_backend(backend)
    await self._pool.put(backend)
    self._semaphore.release()

discard_backend async

discard_backend(backend: AbstractBackend) -> None

Close a broken backend and release its slot.

Use this when a backend failed to launch or is in an unknown state and must not be reused.

Parameters:

Name	Type	Description	Default
backend	AbstractBackend	The backend instance to discard.	required

Source code in wavexis/serve.py

async def discard_backend(self, backend: AbstractBackend) -> None:
    """Close a broken backend and release its slot.

    Use this when a backend failed to launch or is in an unknown state
    and must not be reused.

    Args:
        backend: The backend instance to discard.
    """
    with contextlib.suppress(Exception):
        await backend.close()
    async with self._lock:
        self._created -= 1
    self._semaphore.release()

close_all async

close_all() -> None

Close all pooled backends, drain the pool, and reset all slots.

Source code in wavexis/serve.py

async def close_all(self) -> None:
    """Close all pooled backends, drain the pool, and reset all slots."""
    while not self._pool.empty():
        backend = self._pool.get_nowait()
        with contextlib.suppress(Exception):
            await backend.close()
    async with self._lock:
        self._created = 0
    # Reset the semaphore so leftover acquired slots do not leak between
    # lifecycles (e.g., across tests).
    self._semaphore = asyncio.Semaphore(self._max_concurrent)

set_allowed_base_dir

set_allowed_base_dir(path: str | None) -> None

Set the base directory that serve-mode file paths must be inside of.

Parameters:

Name	Type	Description	Default
path	str | None	Absolute path to the allowed base directory, or None to allow any path (default, not recommended for production).	required

Source code in wavexis/serve.py

def set_allowed_base_dir(path: str | None) -> None:
    """Set the base directory that serve-mode file paths must be inside of.

    Args:
        path: Absolute path to the allowed base directory, or None to allow
            any path (default, not recommended for production).
    """
    global _ALLOWED_BASE_DIR
    _ALLOWED_BASE_DIR = Path(path).resolve() if path else None
    _output_set_allowed_base_dir(path)

with_backend

with_backend(launch_options: BrowserOptions | None = None) -> Callable[[Callable[..., Any]], Callable[[Any], Any]]

Decorator that manages backend lifecycle for serve handlers.

Acquires a backend from the pool, launches it, calls the handler with the backend, and ensures cleanup in a finally block.

Parameters:

Name	Type	Description	Default
launch_options	BrowserOptions | None	BrowserOptions to pass to launch(). Defaults to a plain BrowserOptions().	None

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[[Any], Any]]	A decorator function.

Source code in wavexis/serve.py

def with_backend(
    launch_options: BrowserOptions | None = None,
) -> Callable[[Callable[..., Any]], Callable[[Any], Any]]:
    """Decorator that manages backend lifecycle for serve handlers.

    Acquires a backend from the pool, launches it, calls the handler
    with the backend, and ensures cleanup in a finally block.

    Args:
        launch_options: BrowserOptions to pass to launch(). Defaults to
            a plain BrowserOptions().

    Returns:
        A decorator function.
    """

    def decorator(handler: Any) -> Any:
        async def wrapper(request: Any) -> Any:
            web = _import_aiohttp()
            opts = launch_options or BrowserOptions()
            pool = _get_pool(request)
            backend: AbstractBackend | None = None
            launched = False
            try:
                backend = await pool.get_backend(request.app.get("backend_name"))
                await backend.launch(opts)
                launched = True
                return await handler(request, backend)
            except web.HTTPException:
                raise
            except ActionError as exc:
                return web.json_response(
                    {"error": str(exc)},
                    status=400,
                )
            except WavexisError as exc:
                logger.error("WavexisError in %s: %s", handler.__name__, exc)
                return web.json_response(
                    {"error": "internal server error"},
                    status=500,
                )
            except (json.JSONDecodeError, ValueError, TypeError) as exc:
                # Let plain validation errors from _get_json_body / _safe_params
                # propagate to _json_error_middleware, which returns 400.
                if isinstance(exc, WavexisError):
                    logger.error("WavexisError in %s: %s", handler.__name__, exc)
                    return web.json_response(
                        {"error": "internal server error"},
                        status=500,
                    )
                raise
            except Exception as exc:
                logger.exception("Unhandled error in %s: %s", handler.__name__, exc)
                return web.json_response(
                    {"error": "internal server error"},
                    status=500,
                )
            finally:
                if backend is not None:
                    if launched:
                        await pool.return_backend(backend)
                    else:
                        await pool.discard_backend(backend)

        return wrapper

    return decorator

handle_screenshot async

handle_screenshot(request: Any) -> Any

Handle POST /screenshot — return PNG bytes.

Source code in wavexis/serve.py

async def handle_screenshot(request: Any) -> Any:
    """Handle POST /screenshot — return PNG bytes."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(ScreenshotParams, data)
    from wavexis.actions.screenshot import ScreenshotAction

    action = ScreenshotAction(params)
    image_bytes = await _run_action(request, action)
    return web.Response(body=image_bytes, content_type="image/png")

handle_pdf async

handle_pdf(request: Any) -> Any

Handle POST /pdf — return PDF bytes.

Source code in wavexis/serve.py

async def handle_pdf(request: Any) -> Any:
    """Handle POST /pdf — return PDF bytes."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(PDFParams, data)
    from wavexis.actions.pdf import PDFAction

    action = PDFAction(params)
    pdf_bytes = await _run_action(request, action)
    return web.Response(body=pdf_bytes, content_type="application/pdf")

handle_eval async

handle_eval(request: Any) -> Any

Handle POST /eval — return JSON result.

Source code in wavexis/serve.py

async def handle_eval(request: Any) -> Any:
    """Handle POST /eval — return JSON result."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(EvalParams, data)
    from wavexis.actions.eval import EvalAction

    action = EvalAction(params)
    result = await _run_action(request, action)
    return web.json_response({"result": result})

handle_scrape async

handle_scrape(request: Any) -> Any

Handle POST /scrape — return JSON or CSV.

Source code in wavexis/serve.py

async def handle_scrape(request: Any) -> Any:
    """Handle POST /scrape — return JSON or CSV."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(ScrapeParams, data)
    from wavexis.actions.scrape import ScrapeAction

    action = ScrapeAction(params)
    result = await _run_action(request, action)
    if params.output_format == "csv" and isinstance(result, list) and result:
        import csv
        import io

        buf = io.StringIO()
        writer = csv.DictWriter(buf, fieldnames=result[0].keys())
        writer.writeheader()
        for row in result:
            writer.writerow(row)
        return web.Response(body=buf.getvalue(), content_type="text/csv")
    return web.json_response({"result": result})

handle_dom_get async

handle_dom_get(request: Any) -> Any

Handle POST /dom/get — return HTML as JSON.

Source code in wavexis/serve.py

async def handle_dom_get(request: Any) -> Any:
    """Handle POST /dom/get — return HTML as JSON."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(DOMParams, data)
    from wavexis.actions.dom import DOMAction

    action = DOMAction(params)
    result = await _run_action(request, action)
    return web.json_response({"result": result})

handle_dom_query async

handle_dom_query(request: Any) -> Any

Handle POST /dom/query — return elements as JSON.

Source code in wavexis/serve.py

async def handle_dom_query(request: Any) -> Any:
    """Handle POST /dom/query — return elements as JSON."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(DOMParams, data)
    from wavexis.actions.dom import DOMAction

    action = DOMAction(params)
    result = await _run_action(request, action)
    return web.json_response({"result": result})

handle_navigate async

handle_navigate(request: Any, backend: AbstractBackend) -> Any

Handle POST /navigate — navigate and return status.

Source code in wavexis/serve.py

@with_backend()
async def handle_navigate(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /navigate — navigate and return status."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    url = _get_url_or_400(data.get("url", ""))
    wait_for = data.get("wait_for")
    strategy = (
        WaitStrategy(strategy="selector", selector=wait_for)
        if wait_for
        else WaitStrategy(strategy="load")
    )
    await backend.navigate(url, strategy)
    return web.json_response({"status": "ok", "url": url})

handle_har async

handle_har(request: Any) -> Any

Handle POST /har — return HAR data as JSON.

Source code in wavexis/serve.py

async def handle_har(request: Any) -> Any:
    """Handle POST /har — return HAR data as JSON."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(HarParams, data)
    from wavexis.actions.har import HARAction

    action = HARAction(params)
    result = await _run_action(request, action)
    return web.json_response(result)

handle_cookies_get async

handle_cookies_get(request: Any, backend: AbstractBackend) -> Any

Handle POST /cookies/get — return cookies as JSON.

Source code in wavexis/serve.py

@with_backend()
async def handle_cookies_get(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /cookies/get — return cookies as JSON."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    url = data.get("url", "")
    if url:
        url = _get_url_or_400(url)
        await backend.navigate(url, WaitStrategy(strategy="load"))
    cookies = await backend.get_cookies()
    return web.json_response({"cookies": cookies})

handle_cookies_set async

handle_cookies_set(request: Any, backend: AbstractBackend) -> Any

Handle POST /cookies/set — set a cookie and return status.

Source code in wavexis/serve.py

@with_backend()
async def handle_cookies_set(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /cookies/set — set a cookie and return status."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    cookie_data = data.get("cookie", data)
    params = _safe_params(CookieParams, cookie_data)
    await backend.set_cookie(params)
    return web.json_response({"status": "ok"})

handle_input_click async

handle_input_click(request: Any) -> Any

Handle POST /input/click — click an element.

Source code in wavexis/serve.py

async def handle_input_click(request: Any) -> Any:
    """Handle POST /input/click — click an element."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(InputParams, data)
    params.action = "click"
    from wavexis.actions.input import InputAction

    action = InputAction(params)
    await _run_action(request, action)
    return web.json_response({"status": "ok"})

handle_input_type async

handle_input_type(request: Any) -> Any

Handle POST /input/type — type text into an element.

Source code in wavexis/serve.py

async def handle_input_type(request: Any) -> Any:
    """Handle POST /input/type — type text into an element."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    params = _safe_params(InputParams, data)
    params.action = "type"
    from wavexis.actions.input import InputAction

    action = InputAction(params)
    await _run_action(request, action)
    return web.json_response({"status": "ok"})

handle_perf_metrics async

handle_perf_metrics(request: Any) -> Any

Handle POST /perf/metrics — return performance metrics.

Source code in wavexis/serve.py

async def handle_perf_metrics(request: Any) -> Any:
    """Handle POST /perf/metrics — return performance metrics."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    url = _get_url_or_400(data.get("url", ""))
    from wavexis.actions.performance import PerformanceAction, PerformanceParams

    params = _safe_params(PerformanceParams, {"url": url, "action": "metrics"})
    action = PerformanceAction(params)
    result = await _run_action(request, action)
    return web.json_response(result)

handle_perf_trace async

handle_perf_trace(request: Any) -> Any

Handle POST /perf/trace — return performance trace.

Source code in wavexis/serve.py

async def handle_perf_trace(request: Any) -> Any:
    """Handle POST /perf/trace — return performance trace."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    url = _get_url_or_400(data.get("url", ""))
    duration_ms = _get_int_or_400(
        data.get("duration_ms", 3000), "duration_ms", min_value=1
    )
    from wavexis.actions.performance import PerformanceAction, PerformanceParams

    params = _safe_params(
        PerformanceParams,
        {"url": url, "action": "trace", "duration_ms": duration_ms},
    )
    action = PerformanceAction(params)
    result = await _run_action(request, action)
    return web.json_response(result)

handle_health async

handle_health(request: Any) -> Any

Handle GET /health — return health status.

Source code in wavexis/serve.py

async def handle_health(request: Any) -> Any:
    """Handle GET /health — return health status."""
    web = _import_aiohttp()
    return web.json_response({"status": "ok"})

handle_backends async

handle_backends(request: Any) -> Any

Handle GET /backends — return available backends.

Source code in wavexis/serve.py

async def handle_backends(request: Any) -> Any:
    """Handle GET /backends — return available backends."""
    web = _import_aiohttp()
    manager = get_manager()
    available = manager.list_available()
    return web.json_response(
        {
            "cdp": "cdp" in available,
            "bidi": "bidi" in available,
        }
    )

handle_version async

handle_version(request: Any) -> Any

Handle GET /version — return wavexis version.

Source code in wavexis/serve.py

async def handle_version(request: Any) -> Any:
    """Handle GET /version — return wavexis version."""
    web = _import_aiohttp()
    return web.json_response({"version": __version__})

handle_cwv async

handle_cwv(request: Any) -> Any

Handle POST /cwv — measure Core Web Vitals with scoring.

Body: {"url": "...", "observe_ms": 5000, "budgets": {"lcp_ms": 2500}}

Source code in wavexis/serve.py

async def handle_cwv(request: Any) -> Any:
    """Handle POST /cwv — measure Core Web Vitals with scoring.

    Body: {"url": "...", "observe_ms": 5000, "budgets": {"lcp_ms": 2500}}
    """
    web = _import_aiohttp()
    data = await _get_json_body(request)
    from wavexis.actions.core_web_vitals import (
        CoreWebVitalsAction,
        CoreWebVitalsParams,
    )

    params = _safe_params(CoreWebVitalsParams, data)
    action = CoreWebVitalsAction(params)
    result = await _run_action(request, action)
    return web.json_response(result)

handle_auth async

handle_auth(request: Any, backend: AbstractBackend) -> Any

Handle POST /auth — apply auth context and navigate.

Source code in wavexis/serve.py

@with_backend()
async def handle_auth(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /auth — apply auth context and navigate."""
    web = _import_aiohttp()
    from wavexis.auth import apply_auth_context, load_auth_context

    data = await _get_json_body(request)
    try:
        context_path = _validate_path(data.get("context", ""))
    except WavexisError as e:
        return web.json_response({"error": str(e)}, status=400)
    raw_url = data.get("url", "")
    url = _get_url_or_400(raw_url) if raw_url else ""
    try:
        ctx = await asyncio.to_thread(load_auth_context, str(context_path))
    except (json.JSONDecodeError, OSError) as e:
        return web.json_response({"error": f"Failed to load auth context: {e}"}, status=400)
    await apply_auth_context(backend, ctx, url)
    return web.json_response({"status": "ok", "url": url})

handle_user_agent async

handle_user_agent(request: Any, backend: AbstractBackend) -> Any

Handle POST /user-agent — set custom user agent.

Source code in wavexis/serve.py

@with_backend()
async def handle_user_agent(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /user-agent — set custom user agent."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    ua = _get_str_or_400(data.get("user_agent", ""), "user_agent")
    raw_url = data.get("url", "")
    url = _get_url_or_400(raw_url) if raw_url else ""
    await backend.set_user_agent(ua)
    if url:
        await backend.navigate(url, WaitStrategy(strategy="load"))
    return web.json_response({"status": "ok", "user_agent": ua})

handle_headers async

handle_headers(request: Any, backend: AbstractBackend) -> Any

Handle POST /headers — set custom HTTP headers.

Source code in wavexis/serve.py

@with_backend()
async def handle_headers(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /headers — set custom HTTP headers."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    headers = _get_dict_or_400(data.get("headers", {}), "headers")
    raw_url = data.get("url", "")
    url = _get_url_or_400(raw_url) if raw_url else ""
    await backend.set_headers(headers)
    if url:
        await backend.navigate(url, WaitStrategy(strategy="load"))
    return web.json_response({"status": "ok", "headers": headers})

handle_device async

handle_device(request: Any, backend: AbstractBackend) -> Any

Handle POST /device — emulate a device preset.

Source code in wavexis/serve.py

@with_backend()
async def handle_device(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /device — emulate a device preset."""
    web = _import_aiohttp()
    data = await _get_json_body(request)
    device = _get_device_or_400(data.get("device", ""))
    raw_url = data.get("url", "")
    url = _get_url_or_400(raw_url) if raw_url else ""
    await backend.emulate_device(device)
    if url:
        await backend.navigate(url, WaitStrategy(strategy="load"))
    return web.json_response({"status": "ok", "device": device})

handle_modify_request async

handle_modify_request(request: Any, backend: AbstractBackend) -> Any

Handle POST /modify-request — intercept and modify requests in-flight.

{"url": "...", "pattern": "/api/",

"modifications": {"headers": [...], "method": "...", "post_data": "..."}}

Source code in wavexis/serve.py

@with_backend()
async def handle_modify_request(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /modify-request — intercept and modify requests in-flight.

    Body: {"url": "...", "pattern": "*/api/*",
        "modifications": {"headers": [...], "method": "...", "post_data": "..."}}
    """
    web = _import_aiohttp()
    data = await _get_json_body(request)
    raw_url = data.get("url", "")
    url = _get_url_or_400(raw_url) if raw_url else ""
    pattern_input = data.get("pattern", "*")
    modifications = _get_dict_or_400(data.get("modifications", {}), "modifications")

    if isinstance(pattern_input, str):
        pattern: dict[str, Any] = {"urlPattern": pattern_input}
    elif isinstance(pattern_input, dict):
        pattern = pattern_input
    else:
        raise web.HTTPBadRequest(
            text=json.dumps({"error": "pattern must be a string or a JSON object"}),
            content_type="application/json",
        )

    await backend.modify_request(pattern, modifications)
    if url:
        await backend.navigate(url, WaitStrategy(strategy="load"))
    return web.json_response({"status": "ok", "pattern": pattern})

handle_modify_response async

handle_modify_response(request: Any, backend: AbstractBackend) -> Any

Handle POST /modify-response — intercept and modify responses in-flight.

{"url": "...", "pattern": "/api/",

"modifications": {"status": 200, "body": "...", "content_type": "application/json"}}

Source code in wavexis/serve.py

@with_backend()
async def handle_modify_response(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /modify-response — intercept and modify responses in-flight.

    Body: {"url": "...", "pattern": "*/api/*",
        "modifications": {"status": 200, "body": "...", "content_type": "application/json"}}
    """
    web = _import_aiohttp()
    data = await _get_json_body(request)
    raw_url = data.get("url", "")
    url = _get_url_or_400(raw_url) if raw_url else ""
    pattern_input = data.get("pattern", "*")
    modifications = _get_dict_or_400(data.get("modifications", {}), "modifications")

    if isinstance(pattern_input, str):
        pattern = {"urlPattern": pattern_input}
    elif isinstance(pattern_input, dict):
        pattern = pattern_input
    else:
        raise web.HTTPBadRequest(
            text=json.dumps({"error": "pattern must be a string or a JSON object"}),
            content_type="application/json",
        )

    await backend.modify_response(pattern, modifications)
    if url:
        await backend.navigate(url, WaitStrategy(strategy="load"))
    return web.json_response({"status": "ok", "pattern": pattern_input})

handle_multi async

handle_multi(request: Any, backend: AbstractBackend) -> Any

Handle POST /multi — execute multiple actions from YAML.

Source code in wavexis/serve.py

@with_backend(launch_options=BrowserOptions(headless=True))
async def handle_multi(request: Any, backend: AbstractBackend) -> Any:
    """Handle POST /multi — execute multiple actions from YAML."""
    web = _import_aiohttp()
    from wavexis.record import replay_from_yaml

    data = await _get_json_body(request)
    if not isinstance(data, dict):
        return web.json_response({"error": "body must be a JSON object"}, status=400)
    config_value = data.get("config", "")
    if not isinstance(config_value, str) or not config_value:
        return web.json_response({"error": "config must be a non-empty string path"}, status=400)
    try:
        yaml_path = _validate_path(config_value)
    except WavexisError as e:
        return web.json_response({"error": str(e)}, status=400)
    try:
        results = await replay_from_yaml(yaml_path, backend)
    except WavexisError as e:
        return web.json_response({"error": str(e)}, status=400)
    return web.json_response(
        {
            "status": "ok",
            "actions": len(results),
            "results": [len(r) if isinstance(r, bytes) else str(r)[:200] for r in results],
        }
    )

set_ws_max_connections

set_ws_max_connections(max_conn: int) -> None

Set the maximum number of concurrent WebSocket connections.

Parameters:

Name	Type	Description	Default
max_conn	int	Maximum concurrent WebSocket connections allowed.	required

Source code in wavexis/serve.py

def set_ws_max_connections(max_conn: int) -> None:
    """Set the maximum number of concurrent WebSocket connections.

    Args:
        max_conn: Maximum concurrent WebSocket connections allowed.
    """
    global _ws_max_connections
    _ws_max_connections = max_conn

set_ws_max_messages_per_minute

set_ws_max_messages_per_minute(max_messages: int) -> None

Set the maximum number of WebSocket messages per minute per connection.

Parameters:

Name	Type	Description	Default
max_messages	int	Maximum messages per minute allowed per WebSocket connection.	required

Source code in wavexis/serve.py

def set_ws_max_messages_per_minute(max_messages: int) -> None:
    """Set the maximum number of WebSocket messages per minute per connection.

    Args:
        max_messages: Maximum messages per minute allowed per WebSocket connection.
    """
    global _ws_max_messages_per_minute
    _ws_max_messages_per_minute = max_messages

handle_websocket async

handle_websocket(request: Any) -> Any

Handle GET /ws — WebSocket endpoint for real-time streaming.

Client sends a JSON subscribe message

{ "url": "https://example.com", "events": ["screenshot", "console", "navigation"], "interval": 1.0, "format": "png", "quality": 80 }

Server streams events as JSON messages until the client disconnects.

Source code in wavexis/serve.py

async def handle_websocket(request: Any) -> Any:
    """Handle GET /ws — WebSocket endpoint for real-time streaming.

    Client sends a JSON subscribe message:
        {
            "url": "https://example.com",
            "events": ["screenshot", "console", "navigation"],
            "interval": 1.0,
            "format": "png",
            "quality": 80
        }

    Server streams events as JSON messages until the client disconnects.
    """
    web = _import_aiohttp()

    global _ws_connections
    _ws_acquired = False
    try:
        async with _ws_lock:
            if _ws_connections >= _ws_max_connections:
                return web.Response(
                    status=503,
                    text='{"error": "too many websocket connections"}',
                    content_type="application/json",
                )
            _ws_connections += 1
            _ws_acquired = True

        ws = web.WebSocketResponse()
        tasks: list[asyncio.Task[None]] = []
        backend: AbstractBackend | None = None
        backend_launched = False
        subscription_id: str | None = None
        await ws.prepare(request)
        try:
            msg = await ws.receive()
            if msg.type == web.WSMsgType.TEXT:
                config = json.loads(msg.data)
            else:
                await ws.close()
                return ws
        except (json.JSONDecodeError, KeyError, TypeError):
            await ws.close()
            return ws

        if not isinstance(config, dict):
            await ws.send_json(
                {
                    "type": "error",
                    "message": "Invalid config: JSON body must be an object",
                    "timestamp": time.time(),
                }
            )
            await ws.close()
            return ws

        try:
            url = config.get("url", "about:blank")
            events = config.get("events", ["screenshot"])
            interval = float(config.get("interval", 1.0))
            fmt = config.get("format", "png")
            quality = int(config.get("quality", 80))
        except (TypeError, ValueError):
            await ws.send_json(
                {
                    "type": "error",
                    "message": "Invalid config: interval and quality must be numbers",
                    "timestamp": time.time(),
                }
            )
            await ws.close()
            return ws

        if fmt not in {"png", "jpeg"}:
            await ws.send_json(
                {
                    "type": "error",
                    "message": "Invalid config: format must be 'png' or 'jpeg'",
                    "timestamp": time.time(),
                }
            )
            await ws.close()
            return ws
        if not (0 <= quality <= 100):
            await ws.send_json(
                {
                    "type": "error",
                    "message": "Invalid config: quality must be between 0 and 100",
                    "timestamp": time.time(),
                }
            )
            await ws.close()
            return ws
        if interval <= 0 or interval > 3600:
            await ws.send_json(
                {
                    "type": "error",
                    "message": "Invalid config: interval must be between 0 and 3600",
                    "timestamp": time.time(),
                }
            )
            await ws.close()
            return ws

        try:
            backend = await _get_backend(request)
        except Exception as exc:
            logger.warning("Failed to acquire backend for WebSocket: %s", exc)
            await ws.send_json(
                {
                    "type": "error",
                    "message": "Failed to acquire backend",
                    "timestamp": time.time(),
                }
            )
            await ws.close()
            return ws
        try:
            await backend.launch(BrowserOptions())
            backend_launched = True
            await backend.navigate(url, WaitStrategy(strategy="load"))

            await ws.send_json(
                {
                    "type": "ready",
                    "url": url,
                    "events": events,
                    "timestamp": time.time(),
                }
            )

            if "screenshot" in events:
                tasks.append(
                    asyncio.create_task(
                        _stream_screenshots(ws, backend, interval, fmt, quality),
                    )
                )
            if "console" in events:
                tasks.append(
                    asyncio.create_task(
                        _stream_console(ws, backend, max(interval, 0.5)),
                    )
                )
            if "navigation" in events:
                tasks.append(
                    asyncio.create_task(
                        _stream_navigation(ws, backend, max(interval, 0.5)),
                    )
                )
            if "dom_mutation" in events:
                tasks.append(
                    asyncio.create_task(
                        _stream_dom_mutations(ws, backend, max(interval, 0.5)),
                    )
                )
            if "perf_metrics" in events:
                tasks.append(
                    asyncio.create_task(
                        _stream_perf_metrics(ws, backend, max(interval, 1.0)),
                    )
                )

            subscribe_types = [
                e for e in events if e in ("network_request", "network_response", "dialog")
            ]
            if subscribe_types:

                async def _on_event(event: dict[str, Any]) -> None:
                    await ws.send_json(
                        {
                            "type": event.get("type", "event"),
                            "data": event.get("data", {}),
                            "timestamp": time.time(),
                        }
                    )

                subscription_id = await backend.subscribe_events(subscribe_types, _on_event)

            ws_bucket = TokenBucket(capacity=_ws_max_messages_per_minute, refill_period=60.0)

            async for msg in ws:
                if msg.type == web.WSMsgType.TEXT:
                    allowed = await ws_bucket.acquire()
                    if not allowed:
                        await ws.send_json(
                            {
                                "type": "error",
                                "message": "rate limited: too many messages",
                                "timestamp": time.time(),
                            }
                        )
                        continue
                    try:
                        cmd = json.loads(msg.data)
                    except json.JSONDecodeError:
                        continue
                    if not isinstance(cmd, dict):
                        await ws.send_json(
                            {
                                "type": "error",
                                "message": "Invalid command: JSON body must be an object",
                                "timestamp": time.time(),
                            }
                        )
                        continue
                    action = cmd.get("action")
                    try:
                        if action == "navigate":
                            new_url = cmd.get("url", "")
                            _validate_url_scheme(new_url, allow_empty=False)
                            await backend.navigate(new_url, WaitStrategy(strategy="load"))
                            await ws.send_json(
                                {
                                    "type": "navigated",
                                    "url": new_url,
                                    "timestamp": time.time(),
                                }
                            )
                        elif action == "eval":
                            expr = cmd.get("expression", "")
                            if len(expr) > _MAX_EXPRESSION_LENGTH:
                                err_msg = f"expression exceeds {_MAX_EXPRESSION_LENGTH} characters"
                                await ws.send_json(
                                    {
                                        "type": "error",
                                        "message": err_msg,
                                        "timestamp": time.time(),
                                    }
                                )
                                continue
                            result = await backend.eval(expr)
                            await ws.send_json(
                                {
                                    "type": "eval_result",
                                    "result": result,
                                    "timestamp": time.time(),
                                }
                            )
                        elif action == "screenshot":
                            params = ScreenshotParams(url="", format=fmt, quality=quality)
                            img = await backend.screenshot(params)
                            b64 = base64.b64encode(img).decode("ascii")
                            await ws.send_json(
                                {
                                    "type": "screenshot",
                                    "data": b64,
                                    "timestamp": time.time(),
                                }
                            )
                        elif action == "close":
                            break
                    except WavexisError as exc:
                        await ws.send_json(
                            {
                                "type": "error",
                                "message": str(exc),
                                "timestamp": time.time(),
                            }
                        )
                        continue
                elif msg.type in (
                    web.WSMsgType.CLOSE,
                    web.WSMsgType.CLOSING,
                    web.WSMsgType.CLOSED,
                    web.WSMsgType.ERROR,
                ):
                    break
        except WavexisError as exc:
            await ws.send_json(
                {
                    "type": "error",
                    "message": str(exc),
                    "timestamp": time.time(),
                }
            )
        except Exception:
            logger.exception("Unhandled error in WebSocket handler")
            await ws.send_json(
                {
                    "type": "error",
                    "message": "internal server error",
                    "timestamp": time.time(),
                }
            )
        finally:
            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            if subscription_id is not None and backend is not None:
                with contextlib.suppress(Exception):
                    await backend.unsubscribe_events(subscription_id)
            if backend is not None:
                pool = _get_pool(request)
                if backend_launched:
                    await pool.return_backend(backend)
                else:
                    await pool.discard_backend(backend)
            await ws.close()
    finally:
        if _ws_acquired:
            async with _ws_lock:
                _ws_connections = max(0, _ws_connections - 1)

    return ws

handle_plugins async

handle_plugins(request: Any) -> Any

Handle GET /plugins — list discovered plugins.

Source code in wavexis/serve.py

async def handle_plugins(request: Any) -> Any:
    """Handle GET /plugins — list discovered plugins."""
    from wavexis.plugins import get_registry

    registry = get_registry()
    web = _import_aiohttp()
    return web.json_response(
        {
            "actions": registry.list_actions(),
            "backends": registry.list_backends(),
            "middleware": registry.list_middleware(),
        }
    )

create_app

create_app(backend_name: str | None = None, rate_limit: int | None = None, base_dir: str | None = None, api_key: str | None = None, cors_origins: list[str] | None = None, max_concurrent: int = 5, max_request_size: int = 10 * 1024 * 1024) -> Any

Create and configure the aiohttp web application.

Parameters:

Name	Type	Description	Default
backend_name	str | None	Preferred backend name (e.g. "cdp", "bidi"). If None, auto-detects the first available backend.	None
rate_limit	int | None	Max requests per minute (0 or None = no limit).	None
base_dir	str | None	Base directory for validating file paths in requests. If None, file path access is disabled.	None
api_key	str | None	If set, all requests must include this key as a Bearer token or api_key query parameter.	None
cors_origins	list[str] | None	List of allowed CORS origins. Use ["*"] for all.	None
max_concurrent	int	Max number of concurrent browser backends.	5
max_request_size	int	Maximum request body size in bytes (default 10MB).	10 * 1024 * 1024

Returns:

Type	Description
Any	aiohttp.web.Application with all routes registered.

Raises:

Type	Description
WavexisError	If aiohttp is not installed.
BackendNotAvailableError	If no backend is available.

Source code in wavexis/serve.py

def create_app(
    backend_name: str | None = None,
    rate_limit: int | None = None,
    base_dir: str | None = None,
    api_key: str | None = None,
    cors_origins: list[str] | None = None,
    max_concurrent: int = 5,
    max_request_size: int = 10 * 1024 * 1024,
) -> Any:
    """Create and configure the aiohttp web application.

    Args:
        backend_name: Preferred backend name (e.g. "cdp", "bidi").
            If None, auto-detects the first available backend.
        rate_limit: Max requests per minute (0 or None = no limit).
        base_dir: Base directory for validating file paths in requests.
            If None, file path access is disabled.
        api_key: If set, all requests must include this key as a Bearer
            token or ``api_key`` query parameter.
        cors_origins: List of allowed CORS origins. Use ["*"] for all.
        max_concurrent: Max number of concurrent browser backends.
        max_request_size: Maximum request body size in bytes (default 10MB).

    Returns:
        aiohttp.web.Application with all routes registered.

    Raises:
        WavexisError: If aiohttp is not installed.
        BackendNotAvailableError: If no backend is available.
    """
    web = _import_aiohttp()
    from wavexis.plugins import get_registry

    set_allowed_base_dir(base_dir)

    registry = get_registry()
    middlewares: list[Any] = [m.factory(web) for m in registry.middleware]
    middlewares.append(web.middleware(_request_logging_middleware))
    middlewares.append(web.middleware(_json_error_middleware))

    if cors_origins:
        if "*" in cors_origins and api_key:
            logger.warning(
                "CORS is configured to allow all origins ('*') while API key "
                "authentication is enabled. This allows any website to make "
                "authenticated requests. Consider restricting --cors-origins "
                "to specific domains."
            )
        middlewares.append(
            web.middleware(_cors_middleware(cors_origins, allow_credentials=api_key is not None))
        )

    if api_key:
        middlewares.append(web.middleware(_auth_middleware(api_key)))

    if rate_limit and rate_limit > 0:
        bucket = TokenBucket(capacity=rate_limit, refill_period=60.0)
        middlewares.append(web.middleware(_rate_limit_middleware(bucket)))

    app = web.Application(
        middlewares=middlewares,
        client_max_size=max_request_size,
    )
    manager = get_manager()
    app["backend_name"] = backend_name
    app["backends"] = manager.list_available()
    app["backend_pool"] = BackendPool(max_concurrent=max_concurrent)

    app.router.add_post("/screenshot", handle_screenshot)
    app.router.add_post("/pdf", handle_pdf)
    app.router.add_post("/scrape", handle_scrape)
    app.router.add_post("/dom/get", handle_dom_get)
    app.router.add_post("/dom/query", handle_dom_query)
    app.router.add_post("/navigate", handle_navigate)
    app.router.add_post("/har", handle_har)
    app.router.add_post("/cookies/get", handle_cookies_get)
    app.router.add_post("/cookies/set", handle_cookies_set)
    app.router.add_post("/input/click", handle_input_click)
    app.router.add_post("/input/type", handle_input_type)
    app.router.add_post("/perf/metrics", handle_perf_metrics)
    app.router.add_post("/perf/trace", handle_perf_trace)
    app.router.add_post("/cwv", handle_cwv)
    app.router.add_post("/auth", handle_auth)
    app.router.add_post("/user-agent", handle_user_agent)
    app.router.add_post("/headers", handle_headers)
    app.router.add_post("/device", handle_device)
    app.router.add_post("/modify-request", handle_modify_request)
    app.router.add_post("/modify-response", handle_modify_response)
    app.router.add_post("/multi", handle_multi)
    app.router.add_get("/health", handle_health)
    app.router.add_get("/backends", handle_backends)
    app.router.add_get("/version", handle_version)
    app.router.add_get("/plugins", handle_plugins)

    # /eval and /ws allow arbitrary JavaScript execution; require an API key
    # so they are not exposed unauthenticated.
    if api_key:
        app.router.add_post("/eval", handle_eval)
        app.router.add_get("/ws", handle_websocket)
    else:
        logger.warning(
            "/eval and /ws are disabled because no --api-key was provided"
        )
    return app

serve

serve(port: int = 8080, host: str = 'localhost', backend: str | None = None, rate_limit: int | None = None, base_dir: str | None = None, api_key: str | None = None, cors_origins: list[str] | None = None, max_concurrent: int = 5, max_request_size: int = 10 * 1024 * 1024) -> None

Start the wavexis HTTP server.

Parameters:

Name	Type	Description	Default
port	int	Port to listen on (default 8080).	8080
host	str	Host to bind to (default "localhost").	'localhost'
backend	str | None	Preferred backend name (default auto-detect).	None
rate_limit	int | None	Max requests per minute (0 or None = no limit).	None
base_dir	str | None	Base directory for validating file paths in requests.	None
api_key	str | None	If set, all requests must include this key.	None
cors_origins	list[str] | None	List of allowed CORS origins. Use ["*"] for all.	None
max_concurrent	int	Max concurrent browser backends (default 5).	5
max_request_size	int	Maximum request body size in bytes (default 10MB).	10 * 1024 * 1024

Raises:

Type	Description
WavexisError	If aiohttp is not installed.
BackendNotAvailableError	If no backend is available.

Source code in wavexis/serve.py

def serve(
    port: int = 8080,
    host: str = "localhost",
    backend: str | None = None,
    rate_limit: int | None = None,
    base_dir: str | None = None,
    api_key: str | None = None,
    cors_origins: list[str] | None = None,
    max_concurrent: int = 5,
    max_request_size: int = 10 * 1024 * 1024,
) -> None:
    """Start the wavexis HTTP server.

    Args:
        port: Port to listen on (default 8080).
        host: Host to bind to (default "localhost").
        backend: Preferred backend name (default auto-detect).
        rate_limit: Max requests per minute (0 or None = no limit).
        base_dir: Base directory for validating file paths in requests.
        api_key: If set, all requests must include this key.
        cors_origins: List of allowed CORS origins. Use ["*"] for all.
        max_concurrent: Max concurrent browser backends (default 5).
        max_request_size: Maximum request body size in bytes (default 10MB).

    Raises:
        WavexisError: If aiohttp is not installed.
        BackendNotAvailableError: If no backend is available.
    """
    web = _import_aiohttp()
    logging.basicConfig(
        level=logging.INFO,
        format='{"timestamp":"%(asctime)s","level":"%(levelname)s","logger":"%(name)s","message":%(message)s}',
        datefmt="%Y-%m-%dT%H:%M:%S",
    )
    app = create_app(
        backend,
        rate_limit=rate_limit,
        base_dir=base_dir,
        api_key=api_key,
        cors_origins=cors_origins,
        max_concurrent=max_concurrent,
        max_request_size=max_request_size,
    )
    web.run_app(app, host=host, port=port)