Skip to content

feat: Add Webhook Utility Functions, Improve Docs #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 7 commits into from
Aug 22, 2025
Merged

feat: Add Webhook Utility Functions, Improve Docs #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
6f8330d
stash
maticzav Aug 22, 2025
b46ac62
fixes
maticzav Aug 22, 2025
780bcdd
fixes
maticzav Aug 22, 2025
2a5297d
Update test_webhooks.py
maticzav Aug 22, 2025
36d7dfc
fixes
maticzav Aug 22, 2025
9d72a36
Update test_webhooks.py
maticzav Aug 22, 2025
7ddcd16
Update test_webhooks.py
maticzav Aug 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
245 changes: 108 additions & 137 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,65 +26,6 @@ result.done_output

> The full API of this library can be found in [api.md](api.md).

## Async usage

Simply import `AsyncBrowserUse` instead of `BrowserUse` and use `await` with each API call:

```python
import os
import asyncio
from browser_use_sdk import AsyncBrowserUse

client = AsyncBrowserUse(
api_key=os.environ.get("BROWSER_USE_API_KEY"), # This is the default and can be omitted
)


async def main() -> None:
task = await client.tasks.run(
task="Search for the top 10 Hacker News posts and return the title and url.",
)
print(task.done_output)


asyncio.run(main())
```

Functionality between the synchronous and asynchronous clients is otherwise identical.

### With aiohttp

By default, the async client uses `httpx` for HTTP requests. However, for improved concurrency performance you may also use `aiohttp` as the HTTP backend.

You can enable this by installing `aiohttp`:

```sh
# install from PyPI
pip install browser-use-sdk[aiohttp]
```

Then you can enable it by instantiating the client with `http_client=DefaultAioHttpClient()`:

```python
import asyncio
from browser_use_sdk import DefaultAioHttpClient
from browser_use_sdk import AsyncBrowserUse


async def main() -> None:
async with AsyncBrowserUse(
api_key="My API Key",
http_client=DefaultAioHttpClient(),
) as client:
task = await client.tasks.run(
task="Search for the top 10 Hacker News posts and return the title and url.",
)
print(task.done_output)


asyncio.run(main())
```

## Structured Output with Pydantic

Browser Use Python SDK provides first class support for Pydantic models.
Expand Down Expand Up @@ -115,6 +56,10 @@ asyncio.run(main())

## Streaming Updates with Async Iterators

> When presenting a long running task you might want to show updates as they happen.

Browser Use SDK exposes a `.stream` method that lets you subscribe to a sync or an async generator that automatically polls Browser Use Cloud servers and emits a new event when an update happens (e.g., live url becomes available, agent takes a new step, or agent completes the task).

```py
class HackerNewsPost(BaseModel):
title: str
Expand All @@ -132,7 +77,7 @@ async def main() -> None:
structured_output_json=SearchResult,
)

async for update in client.tasks.stream(structured_task.id, structured_output_json=SearchResult):
async for update in client.tasks.stream(task.id, structured_output_json=SearchResult):
if len(update.steps) > 0:
last_step = update.steps[-1]
print(f"{update.status}: {last_step.url} ({last_step.next_goal})")
Expand All @@ -152,6 +97,109 @@ async def main() -> None:
asyncio.run(main())
```

## Verifying Webhook Events

> You can configure Browser Use Cloud to emit Webhook events and process them easily with Browser Use Python SDK.

Browser Use SDK lets you easily verify the signature and structure of the payload you receive in the webhook.

```py
import uvicorn
import os
from browser_use_sdk.lib.webhooks import Webhook, verify_webhook_event_signature

from fastapi import FastAPI, Request, HTTPException

app = FastAPI()

SECRET_KEY = os.environ['SECRET_KEY']

@app.post('/webhook')
async def webhook(request: Request):
body = await request.json()

timestamp = request.headers.get('X-Browser-Use-Timestamp')
signature = request.headers.get('X-Browser-Use-Signature')

verified_webhook: Webhook = verify_webhook_event_signature(
body=body,
timestamp=timestamp,
secret=SECRET_KEY,
expected_signature=signature,
)

if verified_webhook is not None:
print('Webhook received:', verified_webhook)
else:
print('Invalid webhook received')

return {'status': 'success', 'message': 'Webhook received'}

if __name__ == '__main__':
uvicorn.run(app, host='0.0.0.0', port=8080)
```

## Async usage

Simply import `AsyncBrowserUse` instead of `BrowserUse` and use `await` with each API call:

```python
import os
import asyncio
from browser_use_sdk import AsyncBrowserUse

client = AsyncBrowserUse(
api_key=os.environ.get("BROWSER_USE_API_KEY"), # This is the default and can be omitted
)


async def main() -> None:
task = await client.tasks.run(
task="Search for the top 10 Hacker News posts and return the title and url.",
)
print(task.done_output)


asyncio.run(main())
```

Functionality between the synchronous and asynchronous clients is otherwise identical.

### With aiohttp

By default, the async client uses `httpx` for HTTP requests. However, for improved concurrency performance you may also use `aiohttp` as the HTTP backend.

You can enable this by installing `aiohttp`:

```sh
# install from PyPI
pip install browser-use-sdk[aiohttp]
```

Then you can enable it by instantiating the client with `http_client=DefaultAioHttpClient()`:

```python
import asyncio
from browser_use_sdk import DefaultAioHttpClient
from browser_use_sdk import AsyncBrowserUse


async def main() -> None:
async with AsyncBrowserUse(
api_key="My API Key",
http_client=DefaultAioHttpClient(),
) as client:
task = await client.tasks.run(
task="Search for the top 10 Hacker News posts and return the title and url.",
)
print(task.done_output)


asyncio.run(main())
```

## Advanced

## Handling errors

When the library is unable to connect to the API (for example, due to network connection problems or a timeout), a subclass of `browser_use_sdk.APIConnectionError` is raised.
Expand Down Expand Up @@ -247,8 +295,6 @@ On timeout, an `APITimeoutError` is thrown.

Note that requests that time out are [retried twice by default](#retries).

## Advanced

### Logging

We use the standard library [`logging`](https://docs.python.org/3/library/logging.html) module.
Expand Down Expand Up @@ -294,58 +340,6 @@ These methods return an [`APIResponse`](https://github.com/browser-use/browser-u

The async client returns an [`AsyncAPIResponse`](https://github.com/browser-use/browser-use-python/tree/main/src/browser_use_sdk/_response.py) with the same structure, the only difference being `await`able methods for reading the response content.

#### `.with_streaming_response`

The above interface eagerly reads the full response body when you make the request, which may not always be what you want.

To stream the response body, use `.with_streaming_response` instead, which requires a context manager and only reads the response body once you call `.read()`, `.text()`, `.json()`, `.iter_bytes()`, `.iter_text()`, `.iter_lines()` or `.parse()`. In the async client, these are async methods.

```python
with client.tasks.with_streaming_response.create(
task="Search for the top 10 Hacker News posts and return the title and url.",
) as response:
print(response.headers.get("X-My-Header"))

for line in response.iter_lines():
print(line)
```

The context manager is required so that the response will reliably be closed.

### Making custom/undocumented requests

This library is typed for convenient access to the documented API.

If you need to access undocumented endpoints, params, or response properties, the library can still be used.

#### Undocumented endpoints

To make requests to undocumented endpoints, you can make requests using `client.get`, `client.post`, and other
http verbs. Options on the client will be respected (such as retries) when making this request.

```py
import httpx

response = client.post(
"/foo",
cast_to=httpx.Response,
body={"my_param": True},
)

print(response.headers.get("x-foo"))
```

#### Undocumented request params

If you want to explicitly send an extra param, you can do so with the `extra_query`, `extra_body`, and `extra_headers` request
options.

#### Undocumented response properties

To access undocumented response properties, you can access the extra fields like `response.unknown_prop`. You
can also get all the extra fields on the Pydantic model as a dict with
[`response.model_extra`](https://docs.pydantic.dev/latest/api/base_model/#pydantic.BaseModel.model_extra).

### Configuring the HTTP client

You can directly override the [httpx client](https://www.python-httpx.org/api/#client) to customize it for your use case, including:
Expand Down Expand Up @@ -388,29 +382,6 @@ with BrowserUse() as client:
# HTTP client is now closed
```

## Versioning

This package generally follows [SemVer](https://semver.org/spec/v2.0.0.html) conventions, though certain backwards-incompatible changes may be released as minor versions:

1. Changes that only affect static types, without breaking runtime behavior.
2. Changes to library internals which are technically public but not intended or documented for external use. _(Please open a GitHub issue to let us know if you are relying on such internals.)_
3. Changes that we do not expect to impact the vast majority of users in practice.

We take backwards-compatibility seriously and work hard to ensure you can rely on a smooth upgrade experience.

We are keen for your feedback; please open an [issue](https://www.github.com/browser-use/browser-use-python/issues) with questions, bugs, or suggestions.

### Determining the installed version

If you've upgraded to the latest version but aren't seeing any new features you were expecting then your python environment is likely still using an older version.

You can determine the version that is being used at runtime with:

```py
import browser_use_sdk
print(browser_use_sdk.__version__)
```

## Requirements

Python 3.8 or higher.
Expand Down
65 changes: 65 additions & 0 deletions examples/webhooks.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
#!/usr/bin/env -S rye run python

from typing import Any, Dict, Tuple
from datetime import datetime

from browser_use_sdk.lib.webhooks import (
Webhook,
WebhookAgentTaskStatusUpdate,
WebhookAgentTaskStatusUpdatePayload,
create_webhook_signature,
verify_webhook_event_signature,
)

SECRET = "your-webhook-secret-key"


def mock_webhook_event() -> Tuple[Dict[str, Any], str, str]:
"""Mock a webhook event."""

timestamp = datetime.fromisoformat("2023-01-01T00:00:00").isoformat()

payload = WebhookAgentTaskStatusUpdatePayload(
session_id="sess_123",
task_id="task_123",
status="started",
metadata={"progress": 25},
)

signature = create_webhook_signature(
payload=payload.model_dump(),
timestamp=timestamp,
secret=SECRET,
)

evt: Webhook = WebhookAgentTaskStatusUpdate(
type="agent.task.status_update",
timestamp=datetime.fromisoformat("2023-01-01T00:00:00"),
payload=payload,
)

return evt.model_dump(), signature, timestamp


def main() -> None:
"""Demonstrate webhook functionality."""

# NOTE: You'd get the evt and signature from the webhook request body and headers!
evt, signature, timestamp = mock_webhook_event()

verified_webhook = verify_webhook_event_signature(
body=evt,
expected_signature=signature,
timestamp=timestamp,
secret=SECRET,
)

if verified_webhook is None:
print("✗ Webhook signature verification failed")
else:
print("✓ Webhook signature verified successfully")
print(f" Event type: {verified_webhook.type}")


if __name__ == "__main__":
main()
Loading
Loading