<!-- Source: https://owlbrowser.net/python-sdk/ -->
# Python SDK

Async Python client for the Owl Browser HTTP server. One method per browser tool, generated at construction time from the OpenAPI schema bundled in the package.

## Package Info
- Package: `owl-browser`
- Requirements: Python 3.12+
- License: MIT

## Installation

Install from PyPI. Python 3.12+. Runtime dependencies: aiohttp, pyjwt[crypto], cryptography, beautifulsoup4.

```bash
pip install owl-browser
```

## Quick Start

The SDK is async throughout. `async with` calls `connect()` on entry and `close()` on exit. `create_context()` returns the context id as a string, and every other tool takes `context_id` as a keyword argument.

```python
import asyncio
from owl_browser import OwlBrowser, RemoteConfig

async def main():
    config = RemoteConfig(
        url="http://localhost:8080",
        token="your-secret-token",
        api_prefix="",  # direct to http-server; keep the default "/api" behind nginx
    )

    async with OwlBrowser(config) as browser:
        context_id = await browser.create_context()

        await browser.navigate(context_id=context_id, url="https://example.com", wait_until="load")
        await browser.click(context_id=context_id, selector="button#submit")

        screenshot = await browser.screenshot(context_id=context_id)
        text = await browser.extract_text(context_id=context_id, selector="h1")
        print(f"Page title: {text}")

        await browser.close_context(context_id=context_id)

asyncio.run(main())
```

Sync wrappers exist (`connect_sync`, `execute_sync`, `close_sync`) but each one runs `asyncio.run()` on a fresh event loop, so they do not share a connection. Use the async API.

## Calling Tools

At construction the client reads the OpenAPI schema bundled inside the package, not one fetched from the server, and builds one async method per tool with the `browser_` prefix stripped. Tools whose names do not start with `browser_` (such as `http_request`) keep their full name.

```python
# These two are equivalent
await browser.execute("browser_click", context_id=context_id, selector="#submit")
await browser.click(context_id=context_id, selector="#submit")

# tool_name is positional-only, so tools with their own tool_name param still work
await browser.execute(
    "browser_webmcp_call_tool",
    context_id=context_id,
    tool_name="echo",
    input='{"message":"hi"}',
)

# Introspection is synchronous and offline: it reads the bundled schema
browser.list_tools()               # ['browser_navigate', 'browser_click', ...]
browser.list_methods()             # ['navigate', 'click', ...]
browser.has_method("click")        # True
browser.get_tool("browser_click")  # ToolDefinition | None
```

## Long-Running Tasks

An agentic run (`browser_nla`) routinely takes minutes and blows past the server request timeout, dropping the connection while the browser keeps working. Submit it as a background job instead. `run_task()` is the one-call path: it submits, polls, and returns the unwrapped answer.

```python
# browser_nla needs an LLM on the context: the built-in model, or your own
context_id = await browser.create_context(
    render_mode="agent",
    llm_use_builtin=False,
    llm_is_third_party=True,
    llm_endpoint="https://api.openai.com/v1",
    llm_model="gpt-4o",
    llm_api_key=os.environ["OPENAI_API_KEY"],
)
await browser.navigate(context_id=context_id, url="https://books.toscrape.com", wait_until="load")

# timeout and poll_interval are in seconds; extra kwargs pass through to browser_nla
answer = await browser.run_task(
    context_id,
    "Open the product page for 'The Great Railway Bazaar' and report how many copies are in stock.",
    timeout=600.0,
    poll_interval=2.0,
)
print(answer)  # the tool's own result, envelopes already stripped
```

### The Job API

```python
# Use these when you want the job id itself
job_id = await browser.submit_job(
    "browser_nla", context_id=context_id, command="..."
)

job = await browser.get_job(job_id)   # one job as a dict
jobs = await browser.list_jobs()      # live jobs, no result payloads
await browser.cancel_job(job_id)      # request cancellation

result = await browser.wait_for_job(
    job_id, timeout=600.0, poll_interval=2.0
)
```

### Job States

- `queued`: accepted, not started
- `running`: executing
- `done`: finished, result populated
- `failed`: finished, error populated
- `cancelled`: stopped before completion
- `cancelling`: cancel requested on a job already inside a browser call

Cancellation is cooperative. A queued job cancels immediately; a running one reports `cancelling` until the in-flight call returns. Keep polling through it, it is not terminal.

Jobs are plain REST endpoints and exist on the HTTP transport only. A client built with `TransportMode.WEBSOCKET` raises `OwlBrowserError`. Jobs are held in server memory and do not survive a restart.

## Authentication

### Bearer Token
Simple token-based authentication for most use cases.

```python
config = RemoteConfig(
    url="http://localhost:8080",
    token="your-secret-token"
)
```

### JWT Authentication
RS256, key read from disk, re-signed automatically before expiry.

```python
from owl_browser import RemoteConfig, AuthMode, JWTConfig

config = RemoteConfig(
    url="http://localhost:8080",
    auth_mode=AuthMode.JWT,
    jwt=JWTConfig(
        private_key_path="/path/to/private.pem",
        expires_in=3600,
        refresh_threshold=300,
        issuer="my-app",
        subject="user-123"
    )
)
```

## Configuration

`RemoteConfig` is a dataclass. Paths are built as `{url}{api_prefix}/execute/{tool}`, so keep the default `/api` behind nginx and pass an empty string when you talk to the http-server directly on port 8080.

```python
from owl_browser import RemoteConfig, RetryConfig

config = RemoteConfig(
    url="https://your-domain.com",
    token="secret",

    # Connection settings
    timeout=30.0,        # seconds, per request
    max_concurrent=10,
    verify_ssl=True,

    # max_retries is TOTAL attempts, not extra attempts
    retry=RetryConfig(
        max_retries=3,
        initial_delay_ms=100,
        max_delay_ms=10000,
        backoff_multiplier=2.0,
        jitter_factor=0.1,
    ),

    # "" for a direct connection to the http-server
    api_prefix="/api",
)
```

`timeout` applies per request, with one exception: over HTTP a fixed set of slow tools (navigation, waits, content extraction, screenshots, CAPTCHA and the AI tools) uses `max(120.0, timeout * 4)`. Anything slower than that belongs in a job. Retries are HTTP only; 401, 403 and 429 are never retried.

## Error Handling

Every exception derives from `OwlBrowserError`. A failing tool surfaces as `ToolExecutionError`: any status at or above 400, or a 2xx envelope with `success: false`. Note that `ConnectionError` and `TimeoutError` shadow the Python builtins of the same name.

```python
from owl_browser import (
    AuthenticationError,
    ConnectionError,
    OwlBrowserError,
    RateLimitError,
    TimeoutError,
    ToolExecutionError,
)

try:
    await browser.click(context_id=context_id, selector="#missing")
except ToolExecutionError as e:
    print(f"{e.tool_name} failed: {e.message}")
except RateLimitError as e:
    print(f"rate limited, retry after {e.retry_after}s")
except (AuthenticationError, ConnectionError, TimeoutError) as e:
    print(f"transport problem: {e}")
except OwlBrowserError as e:
    print(f"other SDK error: {e}")
```

`owl_browser.exceptions` also defines `ElementNotFoundError`, `NavigationError`, `ContextLimitError` and `ExpectationError`. The SDK never raises these on its own. They come from the `raise_for_action_result(result)` helper, which you call on a tool result when you want a failed action to become an exception.

## Flow Execution

An optional layer on top of the tool client. Runs declarative JSON flows: steps, captured variables, for_each, retries and expectations. Flows are portable between the Python and Node SDKs.

### Running a Flow

```python
from owl_browser import FlowExecutor, OwlBrowser, RemoteConfig

async def run_flow():
    async with OwlBrowser(RemoteConfig(...)) as browser:
        context_id = await browser.create_context()
        executor = FlowExecutor(browser, context_id)

        flow = FlowExecutor.load_flow("test-flows/navigation.json")
        result = await executor.execute(flow)

        if result.success:
            print(f"Flow completed in {result.total_duration_ms:.0f}ms")
        else:
            print(f"Flow failed: {result.error}")

        await browser.close_context(context_id=context_id)
```

### Flow JSON Format

```json
{
  "name": "Navigation Test",
  "steps": [
    {
      "type": "browser_navigate",
      "url": "https://example.com"
    },
    {
      "type": "browser_extract_text",
      "selector": "h1",
      "expected": {
        "contains": "Example"
      }
    }
  ]
}
```