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 or choose an operation family.

Install

pip install beepgrid
# or: uv add beepgrid

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:

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 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, then run:

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.

"""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 and approval recipe.

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.

Read this page as Markdown