Skip to main content

audd-python

Open .md

Recognize music in audio clips, long broadcasts, and live streams from Python — sync or async.

pip install audd
from audd import AudD

audd = AudD("test") # 10 reqs/day; get a real token at dashboard.audd.io
result = audd.recognize("https://audd.tech/example.mp3")
if result:
print(result.artist, "—", result.title)

Authentication

Get your API token at dashboard.audd.io.

The api_token is resolved on construction in this order:

  1. Explicit api_token= argument.
  2. The AUDD_API_TOKEN environment variable.
  3. Otherwise ValueError, with a pointer to dashboard.audd.io.
from audd import AudD

audd = AudD("your-api-token") # explicit
audd = AudD() # reads AUDD_API_TOKEN

The public "test" token is capped at 10 requests/day and works only for standard recognition — fine for hello-worlds, not for production.

For long-running services pulling tokens from a secret manager, rotate without rebuilding the client:

audd.set_api_token("new-token")

set_api_token is thread-safe and asyncio-safe. In-flight requests finish on the previous token; subsequent calls use the new one.

Recognize a clip

recognize() takes a 5–25-second clip and returns a single RecognitionResult on a match, or None when the clip processed but matched nothing.

from audd import AudD

audd = AudD()

# URL — server fetches the audio
result = audd.recognize("https://audd.tech/example.mp3")

# Filesystem path
from pathlib import Path
result = audd.recognize(Path("/clip.mp3"))

# Bytes
result = audd.recognize(open("/clip.mp3", "rb").read())

# File-like (must be seekable for retry)
with open("/clip.mp3", "rb") as f:
result = audd.recognize(f)

if result:
print(f"{result.artist}{result.title} @ {result.timecode}")
print("song page:", result.song_link)
print("cover art:", result.thumbnail_url)

AsyncAudD exposes the same surface — same names, awaitable:

from audd import AsyncAudD

async with AsyncAudD() as audd:
result = await audd.recognize("https://audd.tech/example.mp3")

A RecognitionResult carries artist, title, album, release_date, label, timecode, song_link, isrc / upc (enterprise plans), and helpers — thumbnail_url, streaming_url(provider), streaming_urls(), preview_url(), pretty_print(). Server fields not yet typed surface via result.model_extra. Full reference: github.com/AudDMusic/audd-python#what-you-get-back.

Source-form notes: file paths reopen on each retry, bytes copy on each retry, file-likes record tell() and seek() back — so unseekable streams cannot be retried (buffer to bytes first if you need that). A str that's neither an http(s):// URL nor an existing path raises TypeError.

Process a long audio file

For broadcasts, podcasts, and full sets longer than 25 seconds, use the enterprise endpoint. It chunks the file server-side and returns every match.

matches = audd.recognize_enterprise(
"/path/to/broadcast.mp3",
limit=10, # stop after 10 matches; ALWAYS set this in development
)
for m in matches:
print(f"{m.timecode} {m.artist}{m.title} (score={m.score})")

Other accepted args: skip, every, skip_first_seconds, use_timecode, accurate_offsets, timeout. The endpoint accepts the same source forms as recognize().

Enterprise calls bill per 12 seconds of audio processed

Always set limit= during development. The unbounded default can ingest many hours of audio on a single call.

Each EnterpriseMatch carries score, timecode, artist, title, album, release_date, label, song_link, isrc/upc (enterprise plans), start_offset, end_offset, plus thumbnail_url, streaming_url(provider), streaming_urls(), pretty_print().

Get streaming-service metadata

Pass return_= to populate provider sub-objects on the result. Without it, those fields are None.

result = audd.recognize(
"https://audd.tech/example.mp3",
return_=["apple_music", "spotify"],
)

if result:
if result.apple_music:
print("Apple Music:", result.apple_music.url)
if result.spotify:
print("Spotify URI:", result.spotify.uri)

# Or resolve any provider — direct URL when the metadata block is set,
# else the lis.tn redirect when song_link is on lis.tn.
print(result.streaming_url("spotify"))
print(result.streaming_urls()) # all providers with a resolvable URL
print(result.preview_url()) # 30-second preview, provider terms apply

Valid providers: apple_music, spotify, deezer, napster, musicbrainz. Each provider adds latency.

The market argument (ISO 3166 region code) controls the Apple Music storefront:

result = audd.recognize(url, return_=["apple_music"], market="GB")

For the field shape of each provider sub-object, see the upstream API reference: docs.audd.io and the GitHub README.

Monitor a live audio stream

A stream in AudD is a long-running audio source — typically a radio URL, HLS, Icecast, or SHOUTcast endpoint — that AudD ingests continuously and recognizes against. Each stream is identified by a radio_id you pick.

There are two ways to consume recognition events:

Both modes need a callback URL configured on the account; otherwise events are never delivered. If your consumers will only longpoll — e.g., browser widgets, mobile apps, or scripts that can't expose a public URL — set the account's callback URL to https://audd.tech/empty/ in advance; it's a placeholder URL that discards POSTs. Consumers themselves don't need to configure anything.

Receive callbacks from a stream

The full setup has three steps: configure the callback URL, register a stream, and parse incoming POSTs.

1. Configure the account callback URL

audd.streams.set_callback_url(
"https://your-app.example.com/audd-callback",
return_metadata=["apple_music", "spotify"], # optional; populates each callback
)

return_metadata is serialized into the URL as ?return=apple_music,spotify (the server reads it from the URL). If you've already baked ?return= into the URL string and also pass return_metadata, the SDK raises rather than silently overwriting.

2. Register the stream

radio_id = 1  # any integer you choose — your handle for this stream

audd.streams.add(
url="https://radio.example.com/stream.mp3",
radio_id=radio_id,
)

Other stream methods:

audd.streams.set_url(radio_id=radio_id, url="https://radio.example.com/new.mp3")
audd.streams.delete(radio_id=radio_id)
streams = audd.streams.list() # list[Stream] with radio_id, url, stream_running, longpoll_category

3. Parse incoming POST bodies

Each callback POST body is JSON with either a result block (recognition) or a notification block (lifecycle event). streams.handle_callback(request) reads the body off your framework's request object and parses it; streams.parse_callback(body) is a pure function for bytes / str / dict (no I/O) — useful for queue replay or webhook proxies.

Exactly one of (match, notification) is non-None on success.

FastAPI — canonical recipe

from fastapi import FastAPI, Request
from audd import AsyncAudD

app = FastAPI()
audd = AsyncAudD()

@app.post("/audd-callback")
async def audd_callback(request: Request) -> dict[str, str]:
match, notif = await audd.streams.handle_callback(request)

if match is not None:
song = match.song
print(f"{song.artist}{song.title} (radio={match.radio_id}, score={song.score})")
# persist, queue, fan-out, etc.

elif notif is not None:
# Stream lifecycle: "stream stopped", "can't connect", etc.
print(f"#{notif.notification_code}: {notif.notification_message}")

return {"ok": "true"}

handle_callback duck-types the request: FastAPI/Starlette via await request.body(), aiohttp via await request.read(), Flask via request.get_data(), Django via request.body. It also accepts raw bytes, str, or any object with a .read().

A StreamCallbackMatch carries radio_id, timestamp, play_length, song (the top match), alternatives (rare extra candidates — may have different artist/title from the top match), and raw_response (the full unparsed body). A StreamCallbackSong mirrors a RecognitionResult's metadata: artist, title, score, album, release_date, label, song_link, plus optional provider blocks when return_metadata was set.

A StreamCallbackNotification carries radio_id, stream_running, notification_code, notification_message, and time.

Other frameworks (Flask, Django, aiohttp, etc.) — see the GitHub README.

Poll for stream events (longpoll)

For consumers that can't expose a public callback URL — mobile apps, browser extensions, scripts behind NAT — open a long-poll subscription instead. Matches arrive over a single HTTP connection your process holds open.

Longpoll requires a configured callback URL

If no callback URL is configured for the account, longpoll might never return any events despite successful song recognition. The SDK preflights this on your first longpoll(...) call and raises AudDInvalidRequestError with guidance if it's missing; pass skip_callback_check=True once you've set one.

If you only want longpoll and have no real receiver, set https://audd.tech/empty/ — it's a placeholder URL that discards incoming POSTs:

audd.streams.set_callback_url("https://audd.tech/empty/")
with audd.streams.longpoll(radio_id=1, timeout=30) as poll:
for match in poll.matches:
print(match.song.artist, "—", match.song.title)

radio_id is any integer you choose — your local handle for this stream. The with block manages the background poller; leaving it shuts the subscription down cleanly.

The poll handle exposes three iterables, all driven by a background thread (or asyncio task, for the async client):

  • poll.matchesStreamCallbackMatch objects as the API recognizes songs.
  • poll.notificationsStreamCallbackNotification lifecycle events (stream started/stopped, errors upstream, etc.).
  • poll.errors — single-shot. The first item terminates the subscription; iterate it concurrently with matches if you want to react to terminal failures.

Async, consuming matches and errors concurrently:

import asyncio
from audd import AsyncAudD

async def main() -> None:
async with AsyncAudD() as audd:
poll = await audd.streams.longpoll(radio_id=1, timeout=30)
async with poll:
async def consume_matches() -> None:
async for m in poll.matches:
print(m.song.artist, "—", m.song.title)

async def watch_errors() -> None:
async for err in poll.errors:
print("terminal:", err)
return

await asyncio.gather(consume_matches(), watch_errors())

asyncio.run(main())

since_time (int, optional) replays events newer than the given server timestamp — useful for resuming after a reconnect. timeout is the per-request hold time in seconds (default 50); the SDK reissues requests automatically until you exit the context.

Tokenless consumers

For browser widgets, mobile clients, or embedded devices that shouldn't carry your api_token, the SDK supports a two-process pattern: a server-side process derives an opaque per-stream identifier — the category — and ships only that string to the consumer. The consumer subscribes with the category alone; no api_token ever leaves the server.

Server-side, derive once and send the result to your client:

category = audd.streams.derive_longpoll_category(radio_id=1)
# ship `category` to the browser / app / device

Consumer-side, subscribe with LongpollConsumer (sync) or AsyncLongpollConsumer (async):

from audd import LongpollConsumer

with LongpollConsumer(category=category) as consumer:
with consumer.iterate(timeout=30) as poll:
for match in poll.matches:
print(match.song.artist, "—", match.song.title)

The poll handle is the same shape as the token-holding flow — matches, notifications, errors. The server-side caller (which holds the token) is responsible for ensuring a callback URL is configured before consumers connect.

Add a song to your custom catalog

The custom-catalog endpoint adds songs to a private fingerprint database for your account. After upload, future recognize() calls on the same account can match against your tracks. It is not how you submit audio for music recognition — for that, use recognize() (or recognize_enterprise()).

danger

The custom-catalog endpoint requires special access. Contact api@audd.io to enable it. Calls without access raise AudDCustomCatalogAccessError.

audio_id = 1  # any integer you choose — your reference to the song

audd.custom_catalog.add(audio_id=audio_id, source="https://my.cdn/song.mp3")

The SDK exposes only add — there is no public list or delete. Track audio_id ↔ song mappings yourself. Re-using an audio_id re-fingerprints that slot.

Handle errors

Every error raised by the SDK subclasses AudDError. Server-reported errors subclass AudDAPIError and carry error_code, message, http_status, request_id, requested_params, request_method, branded_message, and raw_response. Network failures raise AudDConnectionError; malformed JSON raises AudDSerializationError.

Idiomatic error handling

from audd import (
AudD,
AudDAuthenticationError,
AudDQuotaError,
AudDRateLimitError,
AudDInvalidAudioError,
AudDAPIError,
AudDConnectionError,
)

audd = AudD()

try:
result = audd.recognize("/path/to/clip.mp3")
except AudDAuthenticationError as e:
raise SystemExit(f"check your token: [#{e.error_code}] {e.message}")
except AudDQuotaError as e:
print(f"out of quota: {e.message}")
except AudDRateLimitError as e:
print(f"rate limited: {e.message}")
except AudDInvalidAudioError as e:
print(f"audio rejected: {e.message}")
except AudDAPIError as e:
print(f"AudD #{e.error_code}: {e.message} (request_id={e.request_id})")
except AudDConnectionError as e:
print(f"network: {e}")

For some codes — 19 (request_blocked, abuse, test_request_scope), 31337 (ip_ban), 903 (token_disabled) — the server populates branded human-readable text alongside the error. The SDK extracts that onto the exception's branded_message attribute rather than surfacing it as a fake recognition match:

from audd import AudDBlockedError

try:
audd.recognize(...)
except AudDBlockedError as e:
print(f"#{e.error_code} {e.message}")
if e.branded_message:
print(f"server brand text: {e.branded_message}")

When the server returns code 51 (deprecated parameter) with a usable result, the SDK emits a DeprecationWarning and returns the result as if the call had succeeded. Code 51 with no result raises AudDInvalidRequestError.

Retry behavior

Each endpoint is classified into one of three retry classes:

  • READ (streams.list, streams.get_callback_url, longpolls) — retries on any httpx.RequestError, plus HTTP 408/429/5xx.
  • RECOGNITION (recognize, recognize_enterprise, advanced.*) — retries only on pre-upload network errors (ConnectError, ConnectTimeout, WriteError) to avoid double-billing for in-progress uploads, plus 5xx.
  • MUTATING (streams.add, streams.set_url, streams.delete, streams.set_callback_url, custom_catalog.add) — retries only on pre-upload errors; 5xx is surfaced because the side effect may have already happened.

Defaults: max_retries=3, backoff_factor=0.5. Backoff is min(backoff_factor * 2**attempt, 30 seconds) with 0.5x..1.5x jitter. Override per client:

audd = AudD(max_retries=5, backoff_factor=1.0)
audd = AudD(max_retries=1) # disable retries

Configuration

Timeouts

Defaults: standard endpoint 30 s connect / 60 s read & write / 30 s pool. Enterprise endpoint uses 60 minutes for read and write because long files take that long to fingerprint server-side. Override per call:

audd.recognize("https://audd.tech/example.mp3", timeout=10.0)

The argument constructs an httpx.Timeout(timeout) (applies to connect, read, write, pool). For finer-grained control, inject a custom HTTP client.

Custom HTTP client

Inject your own httpx.Client (or httpx.AsyncClient) for proxies, custom transports, custom TLS, or HTTP/2:

import httpx
from audd import AudD

http = httpx.Client(
proxy="http://corp-proxy:8080",
verify="/etc/ssl/internal-ca.pem",
)
audd = AudD("your-api-token", httpx_client=http)

The SDK uses the same injected client for both standard and enterprise endpoints, and doesn't close clients it didn't open.

Observability

The SDK uses the audd logger for stdlib logging. For structured per-request observability, pass an on_event hook:

from audd import AudD, AudDEvent

def on_event(event: AudDEvent) -> None:
if event.kind == "response":
print(f"{event.method} -> {event.http_status} in {event.elapsed_ms:.0f}ms")

audd = AudD(on_event=on_event)

AudDEvent is a frozen dataclass with kind ("request", "response", "exception"), method, url, request_id, http_status, elapsed_ms, error_code, and a free-form extras dict. Events never carry the api_token or request body. Hook exceptions are swallowed (logged at DEBUG on the audd logger) so observability cannot break the request path.

Token rotation

Covered under Authenticationaudd.set_api_token("new-token") swaps atomically without aborting in-flight requests.

Retries

Covered under Retry behavior. max_retries= and backoff_factor= on the constructor.

Calling undocumented endpoints

For AudD endpoints not yet wrapped by typed methods on this SDK, hit them by name through audd.advanced.raw_request(method, params). It returns the raw JSON body as a dict, runs through the same retry policy, and raises AudDSerializationError if the response isn't a JSON object. This is the supported path for anything beta or one-off — typed wrappers ship as features stabilize.

body = audd.advanced.raw_request("getLinks", {"audd_id": 12345})

Concurrency and lifecycle

A single AudD is safe to share across threads; a single AsyncAudD is safe to share across asyncio tasks. Both support context-manager use (with / async with) — and close() / aclose() — to release HTTP resources.

with AudD() as audd:
result = audd.recognize("https://audd.tech/example.mp3")

async with AsyncAudD() as audd:
result = await audd.recognize("https://audd.tech/example.mp3")
Cancellation does not refund server-side metering

Cancelling a recognize() mid-flight might still consume a request on the server — the metered work may already be in progress by the time the cancel signal lands.


dashboard.audd.io · HTTP API reference · Other SDKs