# audd-python


<div className="sdk-page-header-links">
  <a className="button button--primary" href="https://github.com/AudDMusic/audd-python" target="_blank" rel="noopener">View on GitHub</a>
  <a className="button button--secondary" href="https://pypi.org/project/audd/" target="_blank" rel="noopener">PyPI</a>
</div>

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

```bash
pip install audd
```

```python
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](https://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](https://dashboard.audd.io).

```python
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:

```python
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.

```python
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:

```python
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](https://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.

```python
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()`.

:::warning 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`.

```python
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:

```python
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](https://github.com/AudDMusic/audd-python#what-you-get-back).

## 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:

- **Callbacks** — AudD POSTs each match (and each lifecycle notification) to a URL you control. See [Receive callbacks from a stream](#receive-callbacks-from-a-stream).
- **Longpoll** — your code polls AudD's `/longpoll/` endpoint and receives matches and notifications synchronously over HTTP. See [Poll for stream events (longpoll)](#poll-for-stream-events-longpoll).

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

```python
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

```python
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:

```python
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

```python
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](https://github.com/AudDMusic/audd-python#streams).

## 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.

:::warning 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:

```python
audd.streams.set_callback_url("https://audd.tech/empty/")
```

:::

```python
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.matches` — `StreamCallbackMatch` objects as the API recognizes songs.
- `poll.notifications` — `StreamCallbackNotification` 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:

```python
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:

```python
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):

```python
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()`).

:::warning

The custom-catalog endpoint requires special access. Contact [api@audd.io](mailto:api@audd.io) to enable it. Calls without access raise `AudDCustomCatalogAccessError`.

:::

```python
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

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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 [Authentication](#authentication) — `audd.set_api_token("new-token")` swaps atomically without aborting in-flight requests.

### Retries

Covered under [Retry behavior](#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.

```python
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.

```python
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")
```

:::warning 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](https://dashboard.audd.io) · [HTTP API reference](/) · [Other SDKs](/sdks)