pilot-status (Python SDK)

Official Python SDK for the Pilot Status public API.

Installation

pip install pilot-status

Quickstart

Create an API key in the dashboard and use it only on the backend.

import os

from pilot_status import PilotStatusClient

client = PilotStatusClient(
    api_key=os.environ["PILOT_STATUS_API_KEY"],
)

accepted = client.messages.send(
    {
        "templateId": "onboarding-test",
        "destinationNumber": "+5511999999999",
        "variables": {"name": "John"},
    }
)

message = client.messages.get(accepted["id"])
print(message["status"])

Replying without marking the chat read (markRead)

By default a send on a Meta Cloud API number shows the "typing" indicator and marks the contact's last inbound message read (blue ticks). Pass "markRead": False to skip both — e.g. an AI agent answers and a human still sees the conversation as unread later. Meta only allows "typing" as part of a read receipt, so the two cannot be separated. Omit the key to use the number's markReadOnSend setting (default True).

client.messages.send(
    {"destinationNumber": "+5511999999999", "text": "On it!", "markRead": False}
)

No effect on Evolution (unofficial) numbers: sending never marks read there, and read-on-arrival is the number's readMessages setting (turn it off for this use case). Note that POST /v1/messages/typing on a Meta number also marks read.

Management (projects, API keys, numbers)

These endpoints create resources within the scope (project + environment) of the current api_key.

Projects

project = client.projects.create(
    {
        "name": "My Project",
        "description": "Optional description",
    }
)

projects = client.projects.list()

API keys

# Regenerate the default key of one number (tenant-scoped). The new usable key
# is returned once — there is no "create another key" concept.
regenerated = client.api_keys.regenerate_number("wn_1")
print(regenerated["key"])  # shown only once

keys = client.api_keys.list()
# With a number-scoped key this returns that number's masked keys (`ApiKeyListItem[]`).
# With a tenant-scoped key it returns a lean per-number list (`NumberApiKeyReveal[]`):
# { numberId, number, displayName, keyId, keyLast4, key, revealable } — `key` is the real usable value when `revealable` is true.

Message history

# Conversation history (both directions, every provider), newest first.
# This is how you read old messages — the webhook does not replay history.
page = client.messages.history(
    start_date="2026-07-01T00:00:00Z",
    end_date="2026-07-31T23:59:59Z",
    page_size=100,
)
print(page["total"], page["messages"][0]["providerTimestamp"])
# Number-scoped key required. Rows outside the PII window come back with
# content/media/peer nulled and "redacted": True — the envelope is kept.

Leads (Meta Lead Ads)

# Leads from Facebook / Instagram Instant Forms of the Pages linked to the key's
# number, newest first by when the person submitted the form.
page = client.leads.list(
    form_id="1234567890",          # also: ad_id=, page_id=
    start_date="2026-10-01T00:00:00Z",
    end_date="2026-10-31T23:59:59Z",
    page_size=100,
)
for lead in page["leads"]:
    print(lead["leadgenId"], lead["fullName"], lead["consentGiven"])

one = client.leads.get(page["leads"][0]["id"])["lead"]
# Number-scoped key required (`chat:read`). `consentGiven` is True / False / None —
# None means unknown, never a refusal. A lead hidden by the number's PII policy comes
# back with "redacted": True, name/e-mail/phone None and `fields` empty; ids and ad
# attribution are kept. `notice` is set when the number is RELAY_ONLY.

Numbers (WhatsApp)

all_numbers = client.numbers.list()
# JSON array: id, instanceName, primaryLink/secondaryLink, apiKeys refs — no upstream tokens

created = client.numbers.create(
    {
        "name": "My WhatsApp",
        "number": "+5511999999999",
    }
)
# created["qrcodeBase64"], created["pairingCode"] (letter code or None)

refreshed = client.numbers.connect(created["instance"]["id"])
# refreshed["qrcodeBase64"], refreshed["pairingCode"]

status = client.numbers.get_status(created["instance"]["id"])
print(status["state"])
# status["stale"] is False for a live reading (with "checkedAt") and True when
# the provider did not answer and this is the last remembered state (with
# "lastKnownAt"). A 503 raises with body["code"]: PROVIDER_NOT_CONFIGURED is
# permanent (stop polling), UPSTREAM_TIMEOUT / UPSTREAM_ERROR are worth retrying.

detail = client.numbers.get(created["instance"]["id"])
print(detail["settings"]["appliesTo"]["advanced"])  # "evolution-go" | "none"

Per-number settings

numbers.update() is a PARTIAL patch: retention policy and/or the settings block. There is no POST /v1/numbers/{id}/settings, and Evolution's syncFullHistory has no equivalent here.

# stop replaying the device's old messages into the webhook on every reconnect
client.numbers.update_settings(number_id, {"webhookHistoricalMessages": False})

# stop WhatsApp Channel (@newsletter) posts from becoming conversations,
# stored messages and webhook deliveries
client.numbers.update_settings(number_id, {"ignoreNewsletters": True})

# Meta numbers: send without "typing" and without marking the chat read, unless
# a send passes "markRead": True
client.numbers.update_settings(number_id, {"markReadOnSend": False})

client.numbers.update(
    number_id,
    {
        "piiMode": "STORE_X_DAYS",
        "piiRetentionDays": 30,
        "settings": {"rejectCall": True, "msgRejectCall": "I don't take calls here"},
    },
)

historyImportEnabled, webhookHistoricalMessages and ignoreNewsletters apply to every provider. ignoreNewsletters is where a WhatsApp Channel post is refused: the provider exposes no @newsletter gate (unlike ignoreGroups, which it applies itself), so the post always arrives and is dropped on our side — before the conversation, the stored message and the webhook. On a Meta number channels never arrive at all. The other six are the Evolution GO advancedSettings, in the GO dialect (ignoreGroups, not the v2 groupsIgnore); passing None resets one to the provider default. On a Meta number they are stored but never applied — that is what settings["appliesTo"]["advanced"] == "none" means.

markReadOnSend (default True) is the default of the per-send markRead flag and only takes effect on Meta Cloud API numbers; it can also be set at creation ("settings": {"markReadOnSend": False} in numbers.create()).

They are also pushed to the number's connected instances, reported in settingsSync ({applied, failed, skipped}). Best-effort: a disconnected instance keeps the persisted value and picks it up on its next provisioning.

Deleting, restoring and purging

delete() is a 30-day soft delete: the device is logged out right away and the number is kept until restorableUntil, then purged for good. purge() is the permanent delete — immediate and irreversible (history, chats, templates and the number's API keys go with it).

removed = client.numbers.delete(number_id)
print(removed["restorableUntil"])  # ISO-8601 deadline for restore()

client.numbers.restore(number_id)  # gives back the number, NOT the connection:
client.numbers.connect(number_id)  # pair again after restoring

# Delete-and-recreate the SAME phone: create() answers 409 NUMBER_SOFT_DELETED
# while the number is in the window, so purge it first.
client.numbers.delete(number_id)
client.numbers.purge(number_id)

Purging saves nothing compared to delete(): on Pay As You Go a deleted number is already out of the next cycle's charge, and on the other plans paid capacity only drops through a scheduled reduction of extra numbers. Use a tenant-scoped key — a key bound to the deleted number stops authenticating once the number is deleted.

Paying the extra-number charge of a number

A number past the workspace's paid capacity (an extra number) is charged when it first connects. If that charge is refused, the number stays connected but sending from it is blocked — messages.send() raises a 409 NUMBER_BLOCKED with reason: "extra_charge_unpaid" — until the charge is paid. Both numbers.list() and numbers.get(id) say so at the top level of each number:

• extraChargeUnpaidSince — ISO timestamp of the block; None when the number is not blocked.
• extraChargeDisconnectAt — ISO timestamp at which the number is disconnected if it stays unpaid (7 days after the block). Always None on a Meta Cloud API number, which is never disconnected for this.

numbers.pay(id) is "pay now" (POST /v1/numbers/{id}/pay, no body). With a card on file it charges it. It needs a tenant-scoped API key with the billing:manage permission. A wallet top-up or a newly saved card also retries the charge on its own, without this call.

from pilot_status import PilotStatusHttpError

blocked = [n for n in client.numbers.list() if n["extraChargeUnpaidSince"] is not None]

for n in blocked:
    try:
        result = client.numbers.pay(n["id"])
        # "paid": charged, the number can send again.
        # "not_required": nothing was owed for it — safe to call again.
        print(n["id"], result["status"])
    except PilotStatusHttpError as e:
        # 402 has no exception class of its own: branch on the status and on body["code"].
        code = e.body.get("code") if isinstance(e.body, dict) else None
        if e.status == 402 and code == "PAYMENT_FAILED":
            # The card was refused. declineCode is the issuer's reason, when it gave one.
            print(e.body.get("declineCode"), e.body.get("proratedTotal"), e.body.get("currency"))
        elif e.status == 402 and code == "INSUFFICIENT_FUNDS":
            pass  # no credits and no usable card: add credits or a card, then pay again
        elif e.status == 409 and code == "PAYMENT_IN_PROGRESS":
            pass  # another payment of this workspace is running: call again in a few seconds
        elif e.status == 502 and code == "PAYMENT_PROVIDER_ERROR":
            pass  # not a refusal: the charge could not be concluded and the server retries it
        else:
            raise  # 404 NUMBER_NOT_FOUND, 403 NUMBER_SCOPE_NOT_ALLOWED / PERMISSION_DENIED

Every class above is a PilotStatusHttpError, so the single except catches them all. The SDK never retries a POST on its own, so pay() sends exactly one request. The body shape is typed as NumberPayErrorBody, the result as NumberPayResult.

Analytics

stats = client.analytics.get_dashboard_stats(tz="America/Sao_Paulo")
print(stats["totalSent"], stats["failureRate"])

Calls (WhatsApp Business Calling)

Voice calls over the /v1/calls* endpoints, on two kinds of numbers:

• Meta Cloud API numbers — signaling-only: initiate/accept carry the SDP (RFC 8866) produced by your WebRTC client, and audio flows directly between the client and WhatsApp. Settings/permissions/pre_accept are Meta-only.
• Web (Pilot Status / unofficial) numbers — call media is handled server-side, so there is no SDP (omit sdp on initiate/accept). No call permission is required before initiate and there is no Meta per-minute billing. Extra media controls: play (stream an audio file into the call) and realtime_session (full-duplex PCM16 WebSocket).

Numbers on any other provider get 400 FEATURE_NOT_SUPPORTED. call_id arguments accept the Pilot Status id (call_...) or the provider call id (Meta wacid... / Evolution GO CallID).

Billing: on Meta numbers, business-initiated calls (BIC) are billed by Meta directly on your WABA — per minute, in 6-second pulses, only when answered; user-initiated calls (UIC) are free. Calls on web numbers have no Meta billing at all. Pilot Status does not charge for calls.

Web numbers are unofficial (QR-paired) WhatsApp sessions — call quality and availability depend on the paired device/session, and heavy automated calling carries the usual unofficial-number ban risk.

# 1. Permission first (required before calling a user)
perm = client.calls.get_permissions("+5511999999999")
if perm["permission"]["status"] == "no_permission":
    client.calls.request_permission("+5511999999999", text="May we call you?")
    # the user's reply arrives as the call.permission_updated webhook

# 2. Start a business-initiated call (sdp = offer from your WebRTC client)
call = client.calls.initiate({"to": "+5511999999999", "sdp": offer_sdp})

# 3. Answer an inbound call (after the call.ringing webhook)
inbound = client.calls.get("wacid.ABGG...", include_sdp=True)
# feed inbound["sdpOffer"] to your WebRTC client, produce the answer, then:
client.calls.accept(inbound["id"], answer_sdp)

# Other controls
client.calls.reject("wacid.ABGG...")
client.calls.terminate("wacid.ABGG...")

# History + settings (settings are Meta-only)
calls = client.calls.list(limit=25)["calls"]
settings = client.calls.get_settings()
client.calls.update_settings({"status": "ENABLED"})

On a web (Pilot Status) number the same flow needs no SDP and no permission step, and you get server-side media controls:

# Start a call (no sdp, no permission step)
call = client.calls.initiate({"to": "+5511999999999"})

# Answer an inbound call (after the call.ringing webhook) — no sdp
client.calls.accept(call["id"])

# Stream an audio file into the active call (.mp3/.wav/.opus by URL;
# queued and played on connect when the call is not active yet)
client.calls.play(call["id"], "https://cdn.example.com/ivr-greeting.mp3")

# Full-duplex realtime audio: returns {wsUrl, token, expiresInSeconds}.
# Connect a WebSocket to wsUrl and exchange RAW binary PCM16 LE frames
# (plain WebSocket transport, NOT WebRTC; token is single-use, ~2 min)
session = client.calls.realtime_session(call["id"], "talk")

Phone Lines

client.phone_lines requires a tenant-scoped API key (a number-scoped key is refused). The verification routes need phone_lines:purchase, which only the workspace OWNER has; the activation audio needs phone_lines:read. Every verification call returns the same VerificationState: { status, canPurchase, verification } with status one of NONE, PENDING_CONTACT, NEEDS_DOCUMENT, IN_REVIEW, APPROVED, REJECTED.

Account verification

import base64
import uuid

state = client.phone_lines.verification.get()
print(state["status"], state["canPurchase"])

with open("contrato-social.pdf", "rb") as f:
    document = "data:application/pdf;base64," + base64.b64encode(f.read()).decode()

# Idempotency-Key is REQUIRED. Generate it once and REUSE THE SAME KEY when you
# retry this exact request (timeout, network error): the server then returns the
# current state with replayed=True instead of submitting twice.
key = str(uuid.uuid4())
result = client.phone_lines.verification.start(
    document_type="CNPJ",                    # "CPF" | "CNPJ"
    document_number="12.345.678/0001-90",
    contact_email="finance@acme.com.br",
    contact_whatsapp="+55 11 99999-8888",    # country + area code
    document=document,                        # data URL: PDF/JPG/PNG/WEBP, <= 10 MB
    idempotency_key=key,
)
result["state"]["status"]    # usually "PENDING_CONTACT": codes went to the required channels
result["replayed"]           # True when the response is a replay (Idempotent-Replayed: true)

# Contact channels in the PATH are lower-case ("whatsapp" | "email");
# in the response body they are UPPERCASE ("EMAIL" | "WHATSAPP").
client.phone_lines.verification.send_contact_code("whatsapp")   # resend (60 s cooldown)
state = client.phone_lines.verification.confirm_contact("whatsapp", "123456")
state = client.phone_lines.verification.confirm_contact("email", "654321")

# When the document came back UNREADABLE/UNCLEAR, or the status is NEEDS_DOCUMENT:
state = client.phone_lines.verification.submit_document(document)

Errors are the usual typed exceptions with body["code"], e.g. ConflictError IDEMPOTENCY_KEY_REUSED (same key, different body) / IDEMPOTENCY_KEY_IN_USE, RateLimitError RATE_LIMITED (10 submissions/hour, retryAfterSeconds) / CONTACT_CODE_COOLDOWN, ValidationError CONTACT_CODE_INVALID / CONTACT_CODE_EXPIRED.

Activation audio

from pilot_status import NotFoundError

try:
    audio = client.phone_lines.get_activation_audio("act_...")
except NotFoundError as e:
    print(e.body["code"])   # ACTIVATION_NOT_FOUND
else:
    audio.content_type      # e.g. "audio/mpeg"
    audio.content_length    # int | None
    audio.file_name         # from Content-Disposition, or None
    with open(audio.file_name or "activation.mp3", "wb") as f:
        f.write(audio.data)

⚠️ The recording speaks the activation code, which is a credential: never log it, never embed it in a public page. audio.data is left out of repr() on purpose.

Phone-line webhooks

Phone-line events come from a separate webhook registry, signed like the numbers webhooks (x-pilot-status-signature, HMAC-SHA256 hex over the raw body) but with the phone-line secret (plwh_...). Envelope: { id, event, createdAt, data }. They are typed as PhoneLineWebhookEvent (phone_line.purchased, audio_received, code_received, code_failed, renewed, payment_failed, suspended, returned, canceled, verification_updated) and are not accepted by parse_customer_webhook. audioUrl (may be None) points to the audio route above — fetch it with your API key.

Webhooks (parse / validation)

from pilot_status import is_unknown_customer_webhook_event, parse_customer_webhook

def handler(payload: dict):
    event = parse_customer_webhook(payload)

    # An event newer than this SDK version: acknowledge it (2xx), never fail on it.
    if is_unknown_customer_webhook_event(event):
        return

    if event["event"] == "message.failed":
        print(event["data"]["errorMessage"])

    if event["event"] == "call.ended":
        # call.* payloads are FLAT (no "data" wrapper)
        print(event["status"], event.get("duration"))

Events this SDK version does not know

The platform adds events, and a webhook subscribed to "*" receives a new one the day it ships — before you upgrade. parse_customer_webhook does not raise on an event name it does not know: it returns {"type": "unknown", "event", "data", "raw"} (UnknownCustomerWebhookEvent) — event and data as received (data is None for a flat event), raw the whole body, nothing validated. Answer it with a 2xx and ignore it; is_unknown_customer_webhook_event(event) tells it apart (no known event has a top-level type).

It still raises WebhookParseError on a body that is not an object with a string event, and on a known event whose fields break its contract. Pass unknown_events="raise" to keep the old behaviour. is_customer_webhook_event stays strict: False for an unknown event name.

Notes:

• Customer webhook payloads do not include: projectSlug, lastMessageId. Optional correlationId (same as HTTP 202 when present) may appear on outbound status events and on message.reply / message.received when correlated to a prior send.
• message.received includes fromMe (boolean).
• message.group is delivered for inbound group messages (includes groupName).
• message.newsletter is delivered for inbound channel messages (JID ending in @newsletter).
• Supported events in the parser: message.sent, message.delivered, message.read, message.failed, message.reply, message.received, message.group, message.newsletter, number.created, number.connected, number.disconnected, number.removed, number.restored, lead.received, call.ringing, call.connected, call.ended, call.missed, call.permission_updated.
• lead.received: data is the whole lead — { leadId, leadgenId, whatsappNumberId, pageId, pageName?, formId, formName?, adId?, adsetId?, campaignId?, adName?, adsetName?, campaignName?, platform?, isOrganic?, createdAt, fullName?, email?, phone?, fields: [{ key, label?, values }], consentGiven?, conversationId? }. Use leadgenId (Meta's lead id) as the idempotency key; leadId is the id for client.leads.get. In fields, label is absent when it could not be resolved; on client.leads.* it is always present, falling back to the key.
• call.* payloads are flat (fields sit next to event, no data wrapper): { event, callId, externalCallId?, direction, status, from, to, timestamp, duration? }. duration (seconds) appears on call.ended only when the call was answered; call.permission_updated has callId/direction None and status NO_PERMISSION | TEMPORARY | PERMANENT.