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

Install

The library, and a package for each service you use:

pip install operate-as operate-as-sheets

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 calendar

Call

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
)
OptionDefaultWhat it does
base_urlOPERATE_AS_BASE_URL, then https://api.operate.asThe API's address, with or without /v1
timeout60Seconds 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 passA 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 response

Sheets 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:

AttributeWhat it is
statusThe service's own HTTP status, such as 200
dataThe service's own response, parsed from JSON
encodingHow data came back: "json", "text" or "base64"
okTrue 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:

ErrorStatusWhen it happens
InvalidRequestError400, 422An input is missing or the wrong type, or the path leaves the service's API
AuthenticationError401The key is wrong or revoked. Also raised by OperateAs() with no key
PaymentRequiredError402The workspace has no plan: choose one in the dashboard
PermissionDeniedError403Outside 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
NotFoundError404No such action, service or client
ReconnectRequiredError409The client hasn't connected this service, or their access ended
NotConnectedErrornoneThe client doesn't have this service connected, so the call wasn't sent. It's a ReconnectRequiredError too
RateLimitError429Your plan allows no more API calls this period
ServiceError502, 503The service didn't answer, or the provider can't renew access right now
APIConnectionErrornoneOperate As couldn't be reached, or the call timed out
SignatureVerificationErrornoneA webhook delivery failed verification

Any other status raises a plain OperateAsError. Every error has:

AttributeWhat it is
detailThe plain-words message, such as Miller Dental's Google access ended. Send them their link to reconnect.
statusThe HTTP status, or None when no answer arrived
bodyThe parsed error body
headersThe response headers
retry_afterRateLimitError 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
PartDetail
payloadbytes, bytearray or str
toleranceHow many seconds old a delivery may be. Default 300; None skips the check
RaisesSignatureVerificationError 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_HEADERThe 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

On this page