> ## Documentation Index
> Fetch the complete documentation index at: https://aircaps.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Code samples

> Copy-paste examples in Python, Node.js and cURL.

## Transcribe a local file

```python Python theme={null}
# pip install requests
import os, sys, time, requests

API = "https://api.aircaps.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AIRCAPS_API_KEY']}"}

def call(method, path, **kw):
    r = requests.request(method, API + path, headers=HEADERS, timeout=60, **kw)
    if r.status_code >= 400:
        raise SystemExit(r.json()["error"]["message"])
    return r.json()

upload = call("POST", "/files/upload-url")
with open(sys.argv[1], "rb") as f:
    requests.put(upload["upload_url"], data=f, timeout=600).raise_for_status()

job = call("POST", "/transcripts", json={"model": "a5s-async-v1", "file_id": upload["id"], "language_code": "en"})
while job["status"] in ("queued", "processing"):
    time.sleep(3)
    job = call("GET", f"/transcripts/{job['id']}")

if job["status"] == "error":
    raise SystemExit(job["error"]["message"])
for u in job["utterances"]:
    print(f"[{u['start']:8.1f}s] {u['speaker']}: {u['text']}")
```

```javascript Node.js theme={null}
// node transcribe.mjs meeting.mp3   (Node 18+)
import { readFile } from "node:fs/promises";

const API = "https://api.aircaps.com/v1";
const headers = { Authorization: `Bearer ${process.env.AIRCAPS_API_KEY}` };

async function call(method, path, body) {
  const r = await fetch(API + path, {
    method,
    headers: body ? { ...headers, "Content-Type": "application/json" } : headers,
    body: body ? JSON.stringify(body) : undefined,
  });
  const data = await r.json();
  if (!r.ok) throw new Error(data.error.message);
  return data;
}

const upload = await call("POST", "/files/upload-url");
const put = await fetch(upload.upload_url, { method: "PUT", body: await readFile(process.argv[2]) });
if (!put.ok) throw new Error(`upload failed: ${put.status}`);

let job = await call("POST", "/transcripts", { model: "a5s-async-v1", file_id: upload.id, language_code: "en" });
while (job.status === "queued" || job.status === "processing") {
  await new Promise((r) => setTimeout(r, 3000));
  job = await call("GET", `/transcripts/${job.id}`);
}
if (job.status === "error") throw new Error(job.error.message);
for (const u of job.utterances) console.log(`[${u.start.toFixed(1)}s] ${u.speaker}: ${u.text}`);
```

```bash cURL theme={null}
# 1. upload URL
curl -X POST https://api.aircaps.com/v1/files/upload-url -H "Authorization: Bearer $AIRCAPS_API_KEY"
# 2. upload the bytes to the returned upload_url
curl -X PUT --upload-file meeting.mp3 "<upload_url>"
# 3. transcribe with the returned id
curl https://api.aircaps.com/v1/transcripts -H "Authorization: Bearer $AIRCAPS_API_KEY" \
  -H "Content-Type: application/json" -d '{"model": "a5s-async-v1", "file_id": "file_...", "language_code": "en"}'
```

## Stream a WAV file

```python theme={null}
# pip install websockets==15.0.1
# AIRCAPS_API_KEY=aircaps_sk_... python stream.py audio.wav   (mono 16 kHz PCM16 WAV)
import asyncio, json, os, sys, time, wave
import websockets

async def main():
    headers = {"Authorization": f"Bearer {os.environ['AIRCAPS_API_KEY']}"}
    async with websockets.connect("wss://api.aircaps.com/v1/realtime", additional_headers=headers) as ws:
        await ws.send(json.dumps({"type": "start", "providers": ["a5sv2"], "encoding": "pcm_s16le",
                                  "sample_rate": 16000, "channels": 1, "language": "en"}))
        while (event := json.loads(await ws.recv()))["type"] != "ready":
            if event["type"] == "error":
                raise RuntimeError(event["message"])

        async def receive():
            final = ""
            async for message in ws:
                event = json.loads(message)
                if event["type"] == "transcript" and event["is_final"]:
                    # text is the whole transcript so far: print only the newly settled part
                    text = event["text"]
                    print((text[len(final):] if text.startswith(final) else text).strip(), flush=True)
                    final = text
                elif event["type"] in ("session_limit", "error"):
                    print(event["message"], flush=True)

        receiver = asyncio.create_task(receive())
        try:
            with wave.open(sys.argv[1], "rb") as audio:
                start, sent = time.monotonic(), 0
                while chunk := audio.readframes(320):          # 20 ms
                    await ws.send(chunk)
                    sent += len(chunk) // 2
                    await asyncio.sleep(max(0, start + sent / 16000 - time.monotonic()))
            await ws.send(json.dumps({"type": "finish"}))
        except websockets.ConnectionClosedOK:
            pass                                               # the server ended the session (session length)
        await receiver

asyncio.run(main())
```

Convert any file to the right WAV format with `ffmpeg -i input.mp3 -ac 1 -ar 16000 -sample_fmt s16 audio.wav`.

## Verify a webhook

```python theme={null}
# pip install flask standardwebhooks
import os
from flask import Flask, request
from standardwebhooks.webhooks import Webhook, WebhookVerificationError

app = Flask(__name__)
wh = Webhook(os.environ["AIRCAPS_WEBHOOK_SECRET"])

@app.post("/hooks/aircaps")
def hook():
    try:
        event = wh.verify(request.get_data(), dict(request.headers))
    except WebhookVerificationError:
        return "", 400
    print(event["event"], event["id"])     # fetch the transcript with GET /v1/transcripts/{id}
    return "", 204
```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.