Transcribe a call in the background

Voiceasynchronous call transcription

Start or stop background transcription on a live call through REST and receive available transcript text at your status webhook.

transcription

The claim

The vendored REST spec has a calling.transcribe variant of the call command. Its params require control_id and take a status_url, “An HTTP or HTTPS URL that receives the status callback when the transcription completes”. calling.transcribe.stop takes the same control_id. The spec also documents what may arrive: the transcribe status callback, whose event_type is calling.transcript.completed or calling.transcript.failed. Its params.text is “The transcribed text of the call. Omitted when there is no transcribed text.“ The spec calls the callback advisory and best-effort. You reach the two commands as client.calling.transcribe and client.calling.transcribe_stop.

How it works

def start(call_id):
    return client.calling.transcribe(call_id, control_id=CONTROL_ID, status_url=STATUS_URL)

def record(event):
    params = event["params"]
    TRANSCRIPTS[params["call_id"]] = {
        "status": event["event_type"].rsplit(".", 1)[-1],   # completed | failed
        "text": params.get("text"),
        "at": event["timestamp"],
    }

What the platform receives, then what it sends you:

POST /api/calling/calls
{"command": "calling.transcribe", "id": "<call_id>",
 "params": {"control_id": "call-transcript", "status_url": "https://<your-host>/transcripts"}}

POST https://<your-host>/transcripts
{"event_type": "calling.transcript.completed", "timestamp": 1788350400.5,
 "project_id": "...", "space_id": "...",
 "params": {"id": "...", "call_id": "<call_id>", "segment_id": "...", "text": "Hi, this is ..."}}

record reads text with .get, because the spec omits it when nothing was transcribed, whichever status the event carries; the verifier’s failed fixture omits it. Before record runs, the route checks SignalWire’s signature over the request, hex(HMAC(signing_key, url + raw_body)), as signalwire.com/docs/swml/guides/webhook-security documents it and verify-a-webhook-signature explains, so a forged callback cannot overwrite a transcript. GET /transcripts/<call_id> serves what arrived, or pending until something does, to a caller with your READ_TOKEN as a bearer token.

Limitations

You prove the requests and the handler against the documented shapes. What the platform transcribes, and when the callback arrives, are the platform’s side of a live call. The spec calls the callback best-effort, so a transcript that never arrives is a case your job has to expect.

SWML has a transcribe verb for the same job at document time, but it is absent from the 3.0.1 bundled schema. This recipe cannot prove it offline, so it uses the REST command instead.

What to change first

Change record to read params["text"] and run the verifier. The failed callback raises KeyError, because the spec omits the field, which is why the handler reads it with .get.

Where this sits

Before this

Record a call

Handle call status callbacks Reconcile webhooks against the Logs API Verify a webhook signature

python/app.py

"""Transcribe a call in the background.

Two REST commands and one webhook. The vendored REST spec's `calling.transcribe`
variant takes `control_id`, required, and a `status_url`.
`calling.transcribe.stop` takes the same `control_id`. When the transcription
is ready SignalWire may POST the transcribe status callback to `status_url`;
the spec calls it best-effort. Its `event_type` is
`calling.transcript.completed` or `calling.transcript.failed`, and
`params.text` is "The transcribed text of the call. Omitted when there is no
transcribed text."

Nothing here plays into the call or waits on it; the transcript is a document
that may arrive later. The status route accepts a callback only with
SignalWire's signature over it. The read route wants a bearer token of yours,
because a transcript is call content.

Written against signalwire-sdk 3.0.1 (RestClient.calling) and Flask.
"""
import hashlib
import hmac
import os

from dotenv import load_dotenv
from flask import Flask, abort, jsonify, request
from signalwire.rest import RestClient

# the SDK does not read .env for you
load_dotenv()

# RestClient() reads SIGNALWIRE_PROJECT_ID / SIGNALWIRE_API_TOKEN /
# SIGNALWIRE_SPACE from the environment (signalwire/rest/client.py).
client = RestClient()

STATUS_URL = os.getenv("TRANSCRIBE_STATUS_URL")
SIGNING_KEY = os.getenv("SIGNALWIRE_SIGNING_KEY")
READ_TOKEN = os.getenv("READ_TOKEN")
REQUIRED = (("TRANSCRIBE_STATUS_URL", STATUS_URL),
            ("SIGNALWIRE_SIGNING_KEY", SIGNING_KEY),
            ("READ_TOKEN", READ_TOKEN))
for name, value in REQUIRED:
    if not value:
        raise SystemExit(f"{name} is required; see .env.example")

# the platform signs its webhooks: hex(HMAC(signing_key, url + raw_body)), SHA-256
# on call requests, SHA-1 on every signed request (docs/swml/guides/webhook-security)
DIGESTS = {"X-Signalwire-SHA256-Signature": hashlib.sha256,
           "X-Signalwire-Signature": hashlib.sha1}

CONTROL_ID = "call-transcript"

# call_id -> what arrived; swap for your database
TRANSCRIPTS = {}


def start(call_id):
    """Begin transcribing a live call. The callback, if it comes, goes to STATUS_URL."""
    return client.calling.transcribe(call_id, control_id=CONTROL_ID,
                                     status_url=STATUS_URL)


def stop(call_id):
    """Stop the transcription with this control id."""
    return client.calling.transcribe_stop(call_id, control_id=CONTROL_ID)


def signed(headers, url, raw_body):
    """True only when a signature header is present and matches, so a forged
    callback cannot overwrite a transcript."""
    key = SIGNING_KEY
    for header, digest in DIGESTS.items():
        if header in headers:
            sent = headers[header]
            if not all(c in "0123456789abcdefABCDEF" for c in sent):
                return False
            expected = hmac.new(key.encode(), url.encode() + raw_body, digest).hexdigest()
            return hmac.compare_digest(sent, expected)
    return False


def record(event):
    """One status callback. `text` is absent when nothing was transcribed."""
    params = event["params"]
    TRANSCRIPTS[params["call_id"]] = {
        "status": event["event_type"].rsplit(".", 1)[-1],   # completed | failed
        "text": params.get("text"),
        "at": event["timestamp"],
    }


app = Flask(__name__)


@app.post("/transcripts")
def status():
    # SignalWire signs the URL it was given. A configured URL that already
    # carries a query is that URL; one without takes the request's query.
    url = STATUS_URL
    if "?" not in STATUS_URL and request.query_string:
        url += "?" + request.query_string.decode()
    if not signed(request.headers, url, request.get_data()):
        abort(403)
    record(request.get_json(force=True))
    return "", 204


@app.get("/transcripts/<call_id>")
def read(call_id):
    # transcripts are call content; a bearer token of yours gates the read
    if request.headers.get("Authorization") != f"Bearer {READ_TOKEN}":
        abort(401)
    return jsonify(TRANSCRIPTS.get(call_id) or {"status": "pending", "text": None})


if __name__ == "__main__":
    import sys

    if len(sys.argv) == 3 and sys.argv[1] in ("start", "stop"):
        print((start if sys.argv[1] == "start" else stop)(sys.argv[2]))
    elif len(sys.argv) == 1:
        app.run(port=int(os.getenv("PORT", "8080")))
    else:
        raise SystemExit("usage: python app.py            (serve the status URL)\n"
                         "       python app.py start|stop <call_id>")

Run it

cd python
pip install -r requirements.txt
cp ../.env.example .env          # then edit .env: credentials, TRANSCRIBE_STATUS_URL, the signing key, READ_TOKEN
python app.py                    # the status URL, on port 8080

cd python                        # in another shell, during a live call
python app.py start <call_id>
python app.py stop <call_id>     # later, with the same control id

The status URL needs a public HTTPS address. For a local run, expose port 8080 with a tunnel such as ngrok, and set TRANSCRIBE_STATUS_URL to https://<your-host>/transcripts. The call id is the id of a call in progress; handle-call-status-callbacks or the calling logs give you one. When the call ends, read the transcript with your token. Paste the READ_TOKEN value from .env; load_dotenv() loads it into the app, not into your shell:

curl -H "Authorization: Bearer YOUR_READ_TOKEN" "https://YOUR_HOST/transcripts/CALL_ID"

Verify it

No network, no account.

cd ..                     # back to the recipe folder
python verify.py

The verifier swaps the SDK’s HTTP layer for a recorder and drives the Flask app with its test client. It asserts the following.

start and stop each make one POST to the documented calling path, and each body equals one expected object
the spec’s two variants require exactly control_id, and every param sent is documented
the spec’s callback has exactly two event types, and it says text is omitted when there is no transcribed text
two fixture callbacks, one completed with text and one failed without it, carry every required field at both levels and nothing undocumented
the route refuses an unsigned callback, and one signed with another key, with 403, and the store stays empty
a callback signed with the SHA-256 header is accepted; a valid SHA-1 signature beside an empty SHA-256 header is refused, because the header that is present decides
a read without the bearer token is refused with 401
an unknown call reads as pending; after the signed callbacks, the completed call reads back its text and the failed one reads back no text
a status URL configured with its own query string is signed as that URL, once; a signature over the query appended twice is refused

Report an issue