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.
        """
        self._capacity = capacity
        self._tokens = float(capacity)
        self._refill_rate = capacity / refill_period if refill_period > 0 else 0
        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:
            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

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.
    """
    self._capacity = capacity
    self._tokens = float(capacity)
    self._refill_rate = capacity / refill_period if refill_period > 0 else 0
    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:
        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.

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.
    """

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

    async def acquire(self) -> None:
        """Acquire a slot, blocking if the pool is full."""
        await self._semaphore.acquire()

    def release(self) -> None:
        """Release a slot back to the pool."""
        self._semaphore.release()

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

        Reuses an idle backend if available, otherwise creates a new one.

        Args:
            preferred: Preferred backend name for new instances.

        Returns:
            A backend instance (may or may not be launched yet).
        """
        async with self._lock:
            if not self._pool.empty():
                return self._pool.get_nowait()
            self._created += 1
        return await get_manager().select_with_fallback(preferred)

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

        Args:
            backend: The backend instance to return.
        """
        with contextlib.suppress(Exception):
            await backend.close()
        await self._pool.put(backend)

    async def close_all(self) -> None:
        """Close all pooled backends and clear the pool."""
        while not self._pool.empty():
            backend = self._pool.get_nowait()
            with contextlib.suppress(Exception):
                await backend.close()
        async with self._lock:
            self._created = 0

acquire async

acquire() -> None

Acquire a slot, blocking if the pool is full.

Source code in wavexis/serve.py

async def acquire(self) -> None:
    """Acquire a slot, blocking if the pool is full."""
    await self._semaphore.acquire()

release

release() -> None

Release a slot back to the pool.

Source code in wavexis/serve.py

def release(self) -> None:
    """Release a slot back to the pool."""
    self._semaphore.release()

get_backend async

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

Get a backend from the pool or create a new one.

Reuses an idle backend if available, otherwise creates a new one.

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:
    """Get a backend from the pool or create a new one.

    Reuses an idle backend if available, otherwise creates a new one.

    Args:
        preferred: Preferred backend name for new instances.

    Returns:
        A backend instance (may or may not be launched yet).
    """
    async with self._lock:
        if not self._pool.empty():
            return self._pool.get_nowait()
        self._created += 1
    return await get_manager().select_with_fallback(preferred)

return_backend async

return_backend(backend: AbstractBackend) -> None

Return a backend to the pool for reuse.

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.

    Args:
        backend: The backend instance to return.
    """
    with contextlib.suppress(Exception):
        await backend.close()
    await self._pool.put(backend)

close_all async

close_all() -> None

Close all pooled backends and clear the pool.

Source code in wavexis/serve.py

async def close_all(self) -> None:
    """Close all pooled backends and clear the pool."""
    while not self._pool.empty():
        backend = self._pool.get_nowait()
        with contextlib.suppress(Exception):
            await backend.close()
    async with self._lock:
        self._created = 0

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

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()
            backend = await _get_backend(request)
            await backend.launch(opts)
            try:
                return await handler(request, backend)
            except WavexisError as exc:
                return web.json_response(
                    {"error": str(exc)}, status=500,
                )
            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:
                await _release_backend(request, 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 request.json()
    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 request.json()
    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 request.json()
    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 request.json()
    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":
        return web.Response(body=result, 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 request.json()
    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 request.json()
    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 request.json()
    url = 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 request.json()
    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 request.json()
    url = data.get("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 request.json()
    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 request.json()
    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 request.json()
    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 request.json()
    url = data.get("url", "")
    from wavexis.actions.performance import PerformanceAction, PerformanceParams

    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 request.json()
    url = data.get("url", "")
    duration_ms = data.get("duration_ms", 3000)
    from wavexis.actions.performance import PerformanceAction, PerformanceParams

    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()
    from wavexis.actions.core_web_vitals import (
        CoreWebVitalsAction,
        CoreWebVitalsParams,
    )

    data = await request.json()
    url = data.get("url", "")
    observe_ms = int(data.get("observe_ms", 5000))
    budgets = data.get("budgets", {})
    params = CoreWebVitalsParams(
        url=url,
        wait=WaitStrategy(strategy="load"),
        budgets=budgets,
        observe_ms=observe_ms,
    )
    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 request.json()
    context_path = _validate_path(data.get("context", ""))
    url = data.get("url", "")
    ctx = load_auth_context(str(context_path))
    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 request.json()
    ua = data.get("user_agent", "")
    url = data.get("url", "")
    await backend.set_user_agent(ua)
    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 request.json()
    headers = data.get("headers", {})
    url = data.get("url", "")
    await backend.set_headers(headers)
    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 request.json()
    device = data.get("device", "")
    url = data.get("url", "")
    await backend.emulate_device(device)
    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 request.json()
    url = data.get("url", "")
    pattern_input = data.get("pattern", "*")
    modifications = data.get("modifications", {})

    # Build pattern dict - accept either string or dict
    if isinstance(pattern_input, str):
        pattern = {"urlPattern": pattern_input}
    else:
        pattern = pattern_input

    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 request.json()
    url = data.get("url", "")
    pattern = data.get("pattern", "*")
    modifications = data.get("modifications", {})
    await backend.modify_response({"urlPattern": pattern}, modifications)
    if url:
        await backend.navigate(url, WaitStrategy(strategy="load"))
    return web.json_response({"status": "ok", "pattern": pattern})

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 request.json()
    yaml_path = _validate_path(data.get("config", ""))
    results = await replay_from_yaml(yaml_path, backend)
    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

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
    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 = web.WebSocketResponse()
    await ws.prepare(request)

    try:
        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

        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))

        backend = await _get_backend(request)
        try:
            await backend.launch(BrowserOptions())
            await backend.navigate(url, WaitStrategy(strategy="load"))
        except Exception:
            await _release_backend(request, backend)
            raise

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

        tasks: list[asyncio.Task[None]] = []
        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(),
                })

            await backend.subscribe_events(subscribe_types, _on_event)

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

        try:
            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
                    action = cmd.get("action")
                    if action == "navigate":
                        new_url = cmd.get("url", "")
                        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", "")
                        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
                elif msg.type in (web.WSMsgType.CLOSE, web.WSMsgType.CLOSING,
                                   web.WSMsgType.CLOSED, web.WSMsgType.ERROR):
                    break
        finally:
            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            await _release_backend(request, backend)
            await ws.close()
    finally:
        async with _ws_lock:
            _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) -> 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

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,
) -> 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.

    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(_cors_middleware(cors_origins))

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

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

    app = web.Application(middlewares=middlewares)
    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("/eval", handle_eval)
    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("/ws", handle_websocket)
    app.router.add_get("/plugins", handle_plugins)
    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) -> 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

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,
) -> 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).

    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,
    )
    web.run_app(app, host=host, port=port)