Open-source real-time pipeline

A reference pipeline built only on open-source tools: MinIO, Kafka, FFmpeg, Valkey and server-sent events.

This page describes a reference spatial audio pipeline built only from open-source tools. A WAV file dropped into an object storage bucket is checked, sent to Spatial9 and rendered to Eclipsa Audio (IAMF), a binaural headphone version and an MP4. A few minutes later, a browser page plays the binaural result, while the finished files are copied into storage and every state change is recorded in PostgreSQL.

It is for developers and infrastructure teams who want a self-hosted, event-driven pipeline with real-time progress. Everything runs locally with docker compose up -d; only the Spatial9 calls leave your machine, over the open MCP and A2A protocols on HTTPS, so no proprietary SDK is needed. The files are in the examples/open-source folder of the Agent Skills repository at /agents/skills, under the MIT licence.

Architecture

The pipeline is event-driven. MinIO publishes a bucket event to Kafka, a worker consumes it and starts a Spatial9 A2A task, and Spatial9 pushes every state change to a webhook service that updates the database, copies files and streams updates to browsers through Valkey and Server-Sent Events.

 WAV into MinIO --event--> Kafka (audio.uploaded) --> worker.py
                                                        |  FFmpeg check + loudness
                                                        |  MCP create_upload_link + PUT
                                                        |  A2A message/send (spatial_audio)
                                                        v
                                            Spatial9 agent (spatial9.ai/a2a)
                                                        | push on every state change
                                                        v
 browser <-- SSE /events <-- Valkey pub/sub <-- webhook.py --> MinIO immersive bucket
                                                        +--> PostgreSQL tracks + track_events
 Prometheus <-- /metrics (worker, webhook) --> Grafana
 Dagster (dagster_catalog.py): nightly backfill of old masters + daily report

Components

Why both MCP and A2A

The pipeline uses both protocols with the same Spatial9 key and the same limits. MCP is used for single tool calls, such as create_upload_link. A2A is used to hand over a long render and be notified when its state changes. The worker imports the shared clients spatial9_mcp.py and spatial9_a2a.py from examples/enterprise/shared. See MCP server and A2A agent.

Files

Requirements

• Docker with Docker Compose.
• The MinIO client mc.
• A Spatial9 personal key created at /connect.
• A public https address for the webhook service, through a reverse proxy or a tunnel.

Configuration

Copy env.example to .env and fill in every value. Never commit the real .env file.

SPATIAL9_API_KEY=s9_replace_me
A2A_WEBHOOK_TOKEN=replace_with_a_long_random_string
PUBLIC_WEBHOOK_URL=https://your-domain.example/a2a/webhook
MINIO_ROOT_USER=spatial9
MINIO_ROOT_PASSWORD=replace_me_too
POSTGRES_PASSWORD=replace_me_as_well
DATABASE_URL=postgresql://spatial9:replace_me_as_well@postgres:5432/spatial9
KAFKA_BROKERS=kafka:9092
VALKEY_URL=redis://valkey:6379/0
MINIO_ENDPOINT=minio:9000

MinIO is configured in the compose file to publish events from buckets with an event rule to the Kafka topic audio.uploaded.

environment:
  MINIO_NOTIFY_KAFKA_ENABLE_PRIMARY: "on"
  MINIO_NOTIFY_KAFKA_BROKERS_PRIMARY: kafka:9092
  MINIO_NOTIFY_KAFKA_TOPIC_PRIMARY: audio.uploaded

Run it

1. Create a key at /connect, then run cp env.example .env and fill in the values.
2. Make the webhook reachable over https with a reverse proxy or tunnel, and set that address as PUBLIC_WEBHOOK_URL.
3. Start the stack with docker compose up -d.
4. Create the masters bucket and its event rule with the MinIO client, as shown below.
5. Copy a song into the bucket with mc cp my-song.wav local/masters/, open http://localhost:8080 and follow my-song.

mc alias set local http://localhost:9000 "$MINIO_ROOT_USER" "$MINIO_ROOT_PASSWORD"
mc mb local/masters
mc event add local/masters arn:minio:sqs::PRIMARY:kafka --event put

The schema is loaded automatically by PostgreSQL on first start, because the compose file mounts schema.sql into /docker-entrypoint-initdb.d.

Exposed ports

Data model

The worker and the webhook are the only writers. tracks has one row per track and version, with primary key (track_id, version) and a unique index on task_id.

track_events is an append-only audit trail of every state change from any source, with track_id, version, event, detail and at.

Processing steps

Step 1: register every file once

The worker subscribes to audio.uploaded with consumer group spatial9-worker and auto commit disabled. For each MinIO record with a .wav, .flac, .aif or .aiff key, it inserts a pending row with ON CONFLICT DO NOTHING and logs a registered event. It commits the Kafka offset only after the row exists, so an upload is never lost during a restart.

Step 2: check the audio with FFmpeg

Before starting a render, the worker downloads the master and runs these checks.

1. ffprobe must read an audio stream; otherwise the track is rejected as "Not a readable audio file."
2. The duration must be at least 30 seconds; shorter files are rejected with the measured length.
3. ffmpeg with the loudnorm=print_format=json filter measures integrated loudness, stored in loudness_lufs.

Rejected tracks get status = 'rejected', an error and a rejected event.

Step 3: respect the limits and start the task

free_slots() counts tracks that are submitted or working and started events in the last 24 hours, and returns the smaller of 3 - running and 20 - today. Only that many pending tracks are started per loop. For each one, the worker calls create_upload_link, uploads the file with a PUT to direct_upload_url, sends an A2A message/send with skill spatial_audio, and registers the webhook with tasks/pushNotificationConfig/set.

task = a2a.send(
    message_id=f"oss-spatial_audio-{track_id}-v{version}",
    skill="spatial_audio",
    data={"upload_id": link["upload_id"], "title": track_id.replace("-", " ").title(),
          "style": "live_concert", "include_binaural": True},
)
a2a.set_webhook(task["id"], os.environ["PUBLIC_WEBHOOK_URL"], os.environ["A2A_WEBHOOK_TOKEN"])

The messageId is stable per track version, so a retry returns the same task instead of starting a second render.

Real-time delivery

The webhook: trust, then verify

POST /a2a/webhook compares X-A2A-Notification-Token with the configured token using a constant-time comparison and returns 401 on mismatch, or 400 if the body has no task ID. It answers 202 immediately and does the work in a background task. That work never trusts the posted body: it re-reads the task with tasks/get, updates tracks, appends a track_events row, and on completion copies each https file link into the immersive bucket under <track_id>/v<version>/<name>. For failed tasks, it stores the text reason from the task status message.

Server-Sent Events

GET /events/{track_id} opens one text/event-stream connection per viewer. It first sends the current state from PostgreSQL, then relays every message published on the Valkey channel track:<track_id>. A : keep-alive comment every 15 seconds prevents proxies from closing an idle stream. Unknown track IDs return 404, and a valid but unregistered ID receives an event: missing message.

Each update carries state, final, error and files. Browsers receive the temporary Spatial9 links for instant playback, while permanent copies sit in MinIO.

The preview page

GET / serves preview.html, a small page that listens to /events and plays the binaural result as soon as the task completes.

Never miss an update

Every two minutes, the worker looks for tasks that are submitted or working with no update for 5 minutes. It reads each with tasks/get and re-posts the task to its own webhook with the correct token, so one code path handles every state change and a missed push never leaves a track stuck.

Batch backfill with Dagster

The worker handles new uploads in real time. dagster_catalog.py covers scheduled catalog work with two assets in a job named nightly_catalog, scheduled with the cron 0 2 * * *.

Run it locally with pip install dagster dagster-webserver and then dagster dev -f dagster_catalog.py. Backfilled tracks are started by the worker as render slots free up.

Observability

Prometheus scrapes every 15 seconds: the worker on port 9100 and the webhook at /metrics on port 8080.

Limits and safety

• Each key allows 3 renders at once and 20 per day, and agents can create up to 120 A2A tasks per hour. free_slots() enforces the render limits before anything starts.
• Failed renders are refunded, and chorus previews are free.
• Pushes are accepted only with the correct X-A2A-Notification-Token, and the task is always re-read with tasks/get.
• Keys live only in .env or a secret manager; never in code or images.

See Limits and quotas for details.

Alternatives

• Kafka is not required. It provides a durable log, but NATS JetStream or RabbitMQ also work, and a small project can poll the bucket from Dagster on a schedule.
• Local stem separation is optional. You can preview stems locally with the open-source Demucs model, but Spatial9 separates and places stems itself for final immersive renders.

Common mistakes

• Polling in a tight loop: it wastes requests and can hit the 120 tasks per hour limit. Use webhooks with a slow fallback poll.
• Trusting the webhook body: check the token in constant time and re-read the task with tasks/get.
• Starting renders without counting: always check free slots first.
• Generating a new idempotency key on retry: the messageId must be stable per track version.
• Building download links by hand: use only the links Spatial9 returns; they are temporary and signed.
• Putting keys in code or images: keep them in .env or a secret manager such as OpenBao or Vault.

Next steps

• A2A agent: task states, push notification configuration and the notification token.
• Workflow automation: the same webhook pattern in n8n, Zapier and Make.
• Enterprise data platforms: governed recipes for managed data platforms.
• Eclipsa Audio and IAMF: what the pipeline delivers.