GPU Queue

A faster-whisper transcription service with a built-in job queue, so multiple agents can share a single GPU without conflicts. The queue/worker/engine core is transport-agnostic and is exposed through two interchangeable servers — pick the one your clients speak:

Both wrap the same QueueManager, so jobs are still processed one at a time and the response payloads are identical (shared app/service.py).

How it works

submit  ─►  ┌──────────────┐   one worker, one GPU   ┌───────────────┐
            │  job queue   │ ──────────────────────► │ WhisperEngine │
poll    ◄─  └──────────────┘   processed serially     └───────────────┘

A job is enqueued and returns a job_id immediately; you then poll its status until it is done (or error). The single worker drains the queue one job at a time, so a shared GPU never runs two transcriptions at once.

Related MCP server: AssemblyAI MCP Server

Install

pip install -e .        # or: uv sync

This pulls in fastmcp (MCP transport), flask + waitress (HTTP transport), and faster-whisper (the engine).

Running the server

Choose one server (they're independent — neither imports the other):

# MCP transport (streamable-http on :8000)
python server.py
python server.py --port 9000
python server.py --transport stdio        # for stdio MCP clients

# Flask transport (HTTP/JSON on :8001)
python flask_server.py
python flask_server.py --port 9001
python flask_server.py --debug            # Flask dev server with auto-reload

Run a single process only — the queue and job registry live in memory. Multiple threads are fine (waitress uses a thread pool); multiple worker processes are not, as they would each hold a separate, unshared queue.

The Flask server (production waitress path included) prints a startup banner and logs one line per request to stdout, e.g. GET /queue -> 200 (1.2ms).

Engine configuration (both servers)

Read from environment variables:

• WHISPER_MODEL (default distil-large-v3.5)

• WHISPER_DEVICE (default auto)

• WHISPER_COMPUTE_TYPE (default auto)

• WHISPER_MODEL_DIR

HTTP API (Flask transport)

Example session:

curl localhost:8001/ping

# submit
curl -X POST localhost:8001/transcriptions \
  -H 'Content-Type: application/json' \
  -d '{"audio_path":"/abs/path/audio.mp3","args":"--format text"}'
# → {"job_id":"j_ab12cd34","position":1, ...}

# poll until "status":"done"
curl localhost:8001/transcriptions/j_ab12cd34

# inspect the queue / cancel
curl localhost:8001/queue
curl -X POST localhost:8001/transcriptions/j_ab12cd34/cancel

MCP API (MCP transport)

Exposes the same operations as five FastMCP tools:

Quick checks against a running MCP server (uses the fastmcp client):

python test_ping.py                       # ping the server
python test_live.py --audio /path/audio.mp3   # full submit → poll → print transcript

args flags (both transports)

The args string accepts the same faster-whisper CLI flags on either transport, e.g. --format srt, --model large-v3, --language en, --diarize, --word-timestamps, --beam-size 5. See _ARG_MAP in app/queue_manager.py for the full list.

📖 Full HTTP reference: see docs/HTTP_API.md for request/ response schemas, status codes, the job lifecycle, and args flags.

Project layout

app/
  queue_manager.py   queue, worker, engine lifecycle, CLI-arg → kwargs bridge
  whisper/engine.py  faster-whisper wrapper (model stays loaded between jobs)
  models.py          JobRecord dataclass
  service.py         payload-shaping shared by both transports
  main.py / tools.py MCP transport (create_mcp_app + tool definitions)
  flask_app.py       Flask transport (create_flask_app + routes)
server.py            run the MCP server      (:8000)
flask_server.py      run the Flask server    (:8001)

Docker

The bundled Dockerfile / docker-compose.yml run the MCP server (CMD ["python", "server.py"], port 8000) with an NVIDIA GPU reservation. To serve the Flask transport in a container instead, override the command, e.g.:

command: python flask_server.py --port 8000

Tests

pytest -m "not integration"   # unit tests (engine mocked) — covers core, MCP, and Flask
pytest -m integration         # requires faster-whisper + a real model/audio file

• tests/test_queue_manager.py — queue/worker/cancellation logic

• tests/test_service.py — shared payload-shaping

• tests/test_tools.py — MCP transport end-to-end (mocked engine)

• tests/test_flask_app.py — Flask routes end-to-end via the test client