# Python

Install the Python client and connect a running worker.

[PyPI](https://pypi.org/project/beepgrid/) · [Website](https://beepgrid.com) · [Source](https://github.com/beepgrid/beepgrid/tree/main/sdk/python)

Requires **Python 3.10 or later**. The client uses the Python standard library and a synchronous API.




































## Developer reference

Start with the [complete Python tour](https://beepgrid.com/docs/sdks/python/tour.md) or choose an operation family.

- [Client setup](https://beepgrid.com/docs/sdks/python/setup.md)
- [Messages and recipients](https://beepgrid.com/docs/sdks/python/messages.md)
- [Requested replies and decisions](https://beepgrid.com/docs/sdks/python/requests.md)
- [Shared chats and membership](https://beepgrid.com/docs/sdks/python/chats.md)
- [Files and attachments](https://beepgrid.com/docs/sdks/python/files.md)
- [Queries, access and diagnostics](https://beepgrid.com/docs/sdks/python/queries.md)
- [Worker lifecycle](https://beepgrid.com/docs/sdks/python/workers.md)
- [Identity and permissions](https://beepgrid.com/docs/sdks/python/identity.md)
- [Channel preferences](https://beepgrid.com/docs/sdks/python/channels.md)
- [Event streams](https://beepgrid.com/docs/sdks/python/events.md)
- [Bound approvals and execution](https://beepgrid.com/docs/sdks/python/execution.md)
- [OSS Coordinator](https://beepgrid.com/docs/sdks/python/coordinator.md)


## Install

```sh
pip install beepgrid
# or: uv add beepgrid
```

```python
from beepgrid import Client
```

Python 3.10 or later is required. The SDK has no runtime dependencies. The distribution and import are both named `beepgrid`.

## Install from source

From a Beepgrid checkout:

```sh
python3 -m venv .venv
. .venv/bin/activate
python3 -m pip install ./sdk/python
```

The distribution and canonical import are both named `beepgrid`.

## Connect without a source checkout

Use the [published-package worker guide](https://github.com/beepgrid/beepgrid/blob/main/docs/CONNECT-AGENT-RECIPES.md#python) to add a complete worker to your own project. The repository examples below are optional source references.

## Run a worker

Use the protected URL/token-file environment from [Connect an agent](https://beepgrid.com/docs/start/connect-agent.md), then run:

```sh
python3 examples/python-worker/main.py
```

The complete maintained worker is included below. It validates the supplied worker identity, receives ordinary messages and replies in the same conversation. Replace its deterministic reply with your existing agent handler.



```python
"""Use an existing worker credential, or explicitly enroll with BEEPGRID_AGENT_ID."""
import os
from pathlib import Path
from beepgrid import Client

token = os.environ.get("BEEPGRID_TOKEN", "")
if os.environ.get("BEEPGRID_TOKEN_FILE"):
    path = Path(os.environ["BEEPGRID_TOKEN_FILE"])
    if path.stat().st_mode & 0o077:
        raise RuntimeError("Credential file must be owner-only (chmod 600)")
    token = path.read_text().strip()
if not token:
    raise RuntimeError("Set BEEPGRID_TOKEN_FILE or BEEPGRID_TOKEN")
worker = Client(os.environ.get("BEEPGRID_URL", "http://localhost:4010"), token)
agent_id = os.environ.get("BEEPGRID_AGENT_ID", "").strip()
if agent_id:
    worker, session = worker.connect(agent_id, os.environ.get("BEEPGRID_AGENT_NAME", "Python worker"), ["message"])
    identity = {"identity_type": "agent", "agent": session["agent"], "instance_id": session["instance_id"]}
else:
    identity = worker.call("GET", "/api/v1/me")
if identity.get("identity_type") != "agent" or not identity.get("agent", {}).get("id") or not identity.get("instance_id"):
    raise RuntimeError("Without BEEPGRID_AGENT_ID, supply an existing worker credential for an agent connection")
print("Agent:", identity["agent"]["id"], "Instance:", identity["instance_id"], flush=True)

def handle(delivery):
    if delivery["payload"]["type"] == "message":
        message = delivery["payload"]["message"]
        worker.send(conversation_id=message["conversation_id"],
                    body="Python worker received: " + message["body"], key="reply/" + delivery["id"])

worker.run(handle)
```

## Send and request a reply

Once you have a `Client` with the worker's URL and token, `send(target=..., body=..., key=...)` sends a message. `ask(target=..., body=..., key=...)` returns a saved request; `wait(request_id, timeout=300)` waits locally for its final response.

Use `respond(request_id, outcome, version=current_version, answer=answer)` with the current version and an outcome valid for the request. See [the operation reference](https://beepgrid.com/docs/reference/sdk-operations.md) and [approval recipe](https://beepgrid.com/docs/recipes/approval.md).

## Errors and shutdown

`BeepgridError` exposes `code`, `message` and HTTP `status`. A local wait timeout raises `TimeoutError` without cancelling the request. Stop the worker with Ctrl+C or pass a stop callback to `run` when embedding it in an existing process.

`run` acknowledges after your handler returns. Save a result or durable handoff first. Network retries do not recreate your model's in-memory state; [checkpoint it in your runtime](https://beepgrid.com/docs/sdks/reliability.md).