Python
The operate-as library for Python, with clients, actions, pass-through, errors and webhook verification.
operate-as is the official Python library. It needs Python 3.9 or newer and has no dependencies beyond the standard library. Each service's actions come typed in a package of their own, such as operate-as-sheets: install the ones you use.
Get started
Create the client with your key
Keep the key from API & webhooks in an environment variable, never in code you share:
export OPERATE_AS_API_KEY="oa_..."import os
from operate_as import OperateAs
operate = OperateAs(os.environ["OPERATE_AS_API_KEY"])Create one and reuse it: it's safe to share between threads. OperateAs() with no key reads OPERATE_AS_API_KEY itself, and raises AuthenticationError straight away if there's none.
Several keys. Each OperateAs uses the key you give it, so a program can use several keys at once: make one per key. OPERATE_AS_API_KEY is only read when you pass no key. Each reaches only what its key allows.
reporting = OperateAs(os.environ["OPERATE_AS_REPORTING_KEY"]) # reads sheets for the weekly report
bookings = OperateAs(os.environ["OPERATE_AS_BOOKINGS_KEY"]) # writes to one client's calendarCall
from operate_as_sheets import Sheets
miller = operate.client("6f1c2a7e-...") # one of your clients, by id
rows = Sheets(miller).read_rows(spreadsheet_id="1BxiM...", range="A1:D20")
print(rows.status, rows.data["values"])operate.client(client_id) sends nothing. The dashboard shows each client's id, and operate.clients.list() lists them with what each has connected:
for client in operate.clients.list():
print(client.id, client.name, client.services)
# 6f1c2a7e-... Miller Dental ['sheets', 'slack']Options
operate = OperateAs(
os.environ["OPERATE_AS_API_KEY"],
base_url="https://api.operate.as", # or set OPERATE_AS_BASE_URL
timeout=60, # seconds, the default
)| Option | Default | What it does |
|---|---|---|
base_url | OPERATE_AS_BASE_URL, then https://api.operate.as | The API's address, with or without /v1 |
timeout | 60 | Seconds to wait before raising APIConnectionError |
Service packages
Every service with actions has a typed package: operate-as-<service> on PyPI, imported as operate_as_<service>. Its class takes a client and has each action as a method:
from operate_as_ai_inbx import AiInbx
from operate_as_sheets import Sheets
miller = operate.client(client_id)
AiInbx(miller).reply_to_email(
email_id=email_id,
body={"from": "kate@millerdental.com", "html": "<p>Thanks for booking!</p>"},
)
rows = Sheets(miller).read_rows(spreadsheet_id="1BxiM...", range="A1:D20")Inputs are keyword arguments and bodies are TypedDicts, each documented, so your editor completes them and Pyright or mypy flags a missing or misspelled one. An input named like a keyword gets a trailing underscore, such as from_. Each class also has request() for any endpoint of the service's own API.
| You pass | A service the client hasn't connected |
|---|---|
operate.client(client_id) | Raises ReconnectRequiredError from the API |
A client from operate.clients.list() or .get() | Raises NotConnectedError before anything is sent, with what they do have in its message |
Each integration's page in these docs names its package. A package's version moves on when its service's actions change; it works with any operate-as of the same minor version. Without a package, client.service("sheets").run("read_rows", inputs) runs any action.
Actions and pass-through
sheets = Sheets(operate.client(client_id))
rows = sheets.read_rows(spreadsheet_id="1BxiM...", range="A1:D20")
print(rows.status) # 200: the service's own status
print(rows.data["values"]) # the service's own responseSheets comes from its package. operate.actions.run("sheets.read_rows", client_id=client_id, input={...}) runs an action by name instead. Every action is on Actions.
The result
Every action and request return a Result:
| Attribute | What it is |
|---|---|
status | The service's own HTTP status, such as 200 |
data | The service's own response, parsed from JSON |
encoding | How data came back: "json", "text" or "base64" |
ok | True when status is 2xx |
When the service answers with a file, result.encoding is "base64" and data is the file as a base64 string. Decode it with base64.b64decode(result.data). encoding is "json" for parsed JSON and "text" for plain text. See The result.
result = sheets.read_rows(spreadsheet_id="1BxiM...", range="A1:D20")
if not result.ok:
print("Sheets said no:", result.status, result.data)Errors
Every exception is an OperateAsError:
| Error | Status | When it happens |
|---|---|---|
InvalidRequestError | 400, 422 | An input is missing or the wrong type, or the path leaves the service's API |
AuthenticationError | 401 | The key is wrong or revoked. Also raised by OperateAs() with no key |
PaymentRequiredError | 402 | The workspace has no plan: choose one in the dashboard |
PermissionDeniedError | 403 | Outside the key's clients, services or actions, a service the client wasn't asked to connect, outside your plan, or the key is beyond your plan's key allowance, or the client paused the connection or allows only reads |
NotFoundError | 404 | No such action, service or client |
ReconnectRequiredError | 409 | The client hasn't connected this service, or their access ended |
NotConnectedError | none | The client doesn't have this service connected, so the call wasn't sent. It's a ReconnectRequiredError too |
RateLimitError | 429 | Your plan allows no more API calls this period |
ServiceError | 502, 503 | The service didn't answer, or the provider can't renew access right now |
APIConnectionError | none | Operate As couldn't be reached, or the call timed out |
SignatureVerificationError | none | A webhook delivery failed verification |
Any other status raises a plain OperateAsError. Every error has:
| Attribute | What it is |
|---|---|
detail | The plain-words message, such as Miller Dental's Google access ended. Send them their link to reconnect. |
status | The HTTP status, or None when no answer arrived |
body | The parsed error body |
headers | The response headers |
retry_after | RateLimitError only: seconds from Retry-After, or None |
Catch reconnect-required
A client's access ending won't fix itself. Catch it, tell someone, and stop working for that client:
import time
import operate_as
try:
sheets.add_row(spreadsheet_id="1BxiM...", range="A1", values=["2026-10-16", "Jane Doe"])
except operate_as.ReconnectRequiredError as error:
notify_team(f"{client_id}: {error.detail}") # "...Send them their link to reconnect."
pause_workflow(client_id)
except operate_as.RateLimitError as error:
time.sleep(error.retry_after or 60)
except operate_as.ServiceError:
retry_later()
except operate_as.OperateAsError as error:
print(error.status, error.detail)See Retries for what's worth retrying.
Verify webhooks
Webhook.verify checks the Operate-As-Signature header and returns the event as a dict. Pass the raw body, before any JSON parsing:
import os
from flask import Flask, request
from operate_as import SignatureVerificationError, Webhook
app = Flask(__name__)
@app.post("/hooks/operate-as")
def operate_as_hook():
try:
event = Webhook.verify(
request.get_data(), # the raw bytes
request.headers.get("Operate-As-Signature"),
os.environ["OPERATE_AS_WEBHOOK_SECRET"], # whsec_...
)
except SignatureVerificationError:
return "", 400
if event["type"] == "connection.expired":
data = event["data"]
email_client(data["clientName"], data["reconnectUrl"])
return "", 204| Part | Detail |
|---|---|
payload | bytes, bytearray or str |
tolerance | How many seconds old a delivery may be. Default 300; None skips the check |
| Raises | SignatureVerificationError for a missing or malformed header, a wrong signature, an old delivery, or a body that isn't JSON |
Webhook.sign(payload, secret, timestamp=None) | Builds a valid header, for testing your handler |
SIGNATURE_HEADER | The header name |
See Webhooks for the events.
Typing
The package ships type hints and is checked with Pyright in strict mode. Action, ActionInput, ActionPage and Result are frozen dataclasses. Client is one of your clients, with service(name) to act for it; ConnectedClient, from operate.clients, adds name and services. Result.data is Any, so narrow it yourself:
from typing import TypedDict
class ValueRange(TypedDict):
range: str
values: list[list[str]]
rows = sheets.read_rows(spreadsheet_id="1BxiM...", range="A1:D20")
values: ValueRange = rows.data