OfSpectrum Python SDK

Official Python SDK for the OfSpectrum audio watermarking API.

Installation

pip install ofspectrum

Or install from source:

pip install -e /path/to/neo/sdk

Quick Start

from ofspectrum import OfSpectrum

client = OfSpectrum(api_key="your_api_key")

# Create a Standard token.
token = client.tokens.create(name="Production Token")
print(f"Created token: {token.id}")

# Encode and save watermarked audio.
result = client.audio.encode(
    audio="input.mp3",
    token_id=token.id,
)
result.save("watermarked.mp3")
print(f"Encoded {result.audio_duration}s of audio")

# Decode watermark from audio.
decode = client.audio.decode("suspect.mp3")
if decode.watermarked:
    print(f"Watermark detected. Token ID: {decode.token_id}")
else:
    print("No watermark detected")

# Check your quota.
quota = client.quotas.get_encode_quota()
print(f"Remaining encode quota: {quota.remaining}/{quota.limit} seconds")

Building With AI Coding Agents

If you are building an app or internal tool with this SDK, see AGENT_GUIDE.md. It explains token modeling patterns, notebook/provenance guidance, security notes, test flows, and includes a copyable prompt for AI coding agents such as Codex.

Test Audio

Synthetic WAV files for encode/decode smoke tests are available in examples/audio. They contain no third-party audio and are intended for local SDK verification.

Token Management

Standard tokens are the simplest option. Pro tokens support workflow-specific verification-key configuration.

import os

# List all tokens.
tokens = client.tokens.list()

# Get a specific token.
token = client.tokens.get("token-uuid")

# Create a Pro token when your workflow requires a configurable verification key.
verification_key = int(os.environ["OFSPECTRUM_PUBLIC_KEY"])
token = client.tokens.create(
    name="Pro Token",
    token_type="pro",
    public_key=verification_key,
)

# Update a token name.
token = client.tokens.update(
    token_id="token-uuid",
    name="New Name",
)

# Update a Pro token verification key.
token = client.tokens.update(
    token_id="token-uuid",
    public_key=verification_key,
)

# Upgrade an existing Standard token to Pro.
# Token types can be upgraded, but not downgraded.
token = client.tokens.update(
    token_id="token-uuid",
    token_type="pro",
    public_key=verification_key,
)

# Configure how the token may be used by AI systems.
token = client.tokens.update(
    token_id="token-uuid",
    ai_auth_enabled=True,
    ai_auth_access_type="direct_use",  # or "premium_track"
    ai_auth_price=5,
    ai_auth_other_instructions="Attribution required",
    ai_auth_tags=["voice", "licensed"],
)

Passing ai_auth_price=None clears the price. Passing ai_auth_tags=[] removes all AI authorization tags from the token.

Reusable AI authorization tags can be listed or created separately:

tags = client.tokens.list_ai_auth_tags()
voice_tag = client.tokens.create_ai_auth_tag("Voice Clone")

client.tokens.update(
    token_id="token-uuid",
    ai_auth_tags=[voice_tag.tag],
)

Creating a tag does not attach it to a token. Pass the selected tag names to tokens.create() or tokens.update() to associate them with a token.

Token deletion is not available via API. Tokens are consumable resources.

Audio Watermarking

result = client.audio.encode(
    audio="input.mp3",
    token_id=token.id,
    strength=1.0,
    smooth=True,
)
result.save("output.mp3")

decode = client.audio.decode("suspect.mp3")
if decode.watermarked:
    print(f"Token: {decode.token_id}")

Use decode(..., public_key=verification_key) only when your workflow requires an explicit verification key.

Streaming PCM Encode

Use stream_encode_pcm() when your application already works with raw PCM audio or needs low-latency streaming from a file-processing pipeline, microphone, call, meeting, or live stream.

The input must be raw PCM float32 little-endian bytes. 48 kHz mono is recommended. The SDK does not currently decode MP3/WAV/FLAC files, resample audio, or convert containers for this streaming method.

def chunk_pcm(pcm_bytes: bytes, chunk_seconds: float = 0.5):
    sample_rate = 48000
    channels = 1
    bytes_per_second = sample_rate * channels * 4
    chunk_size = int(bytes_per_second * chunk_seconds)
    for offset in range(0, len(pcm_bytes), chunk_size):
        yield pcm_bytes[offset:offset + chunk_size]

result = client.audio.stream_encode_pcm(
    pcm_chunks=chunk_pcm(pcm_f32le_bytes),
    token_id=token.id,
    sample_rate=48000,
    channels=1,
    smooth=True,
)

encoded_pcm = result.encoded_pcm
print(f"Encoded {result.audio_duration:.2f}s of PCM")

encoded_pcm is raw PCM float32 little-endian, not WAV or MP3. Wrap it in a WAV container or encode it to your desired output format before playback or download.

Notebook Management

Attach notes and media files to tokens. Private notebooks require a credential, and limits depend on your account and token configuration.

Notebook limits:

Token Type	Public Notebooks	Private Notebooks
standard	1	1
pro	1	Unlimited
enterprise	1	Unlimited

If the limit is reached, the SDK raises a ValidationError with a customer-facing message.

notebook = client.notebooks.create(
    token_id=token.id,
    note_name="Release Notes",
    text_content="## Version 1.0\n\nRelease notes.",
    is_public=True,
)

private_notebook = client.notebooks.create(
    token_id=token.id,
    note_name="Private Notes",
    text_content="Confidential content",
    is_public=False,
    credential_val="choose-a-secure-credential",
)

client.notebooks.upload_media(
    note_id=notebook.id,
    file="cover.jpg",
)

notebooks = client.notebooks.list(token_id=token.id)

Each notebook accepts up to 10 media files. Each file may be up to 100 MB, and the combined media size limit is 10 GB per notebook.

Quota Checking

quota = client.quotas.get_encode_quota()
print(f"Remaining encode quota: {quota.remaining}/{quota.limit}")

decode_quota = client.quotas.get_decode_quota()
print(f"Remaining decode quota: {decode_quota.remaining}/{decode_quota.limit}")

if client.quotas.check_encode_available(duration_seconds=300):
    result = client.audio.encode(audio="input.mp3", token_id=token.id)

Error Handling

from ofspectrum import (
    OfSpectrumError,
    AuthenticationError,
    RateLimitError,
    QuotaExceededError,
    WatermarkExistsError,
    ResourceNotFoundError,
)

try:
    result = client.audio.encode(audio="input.mp3", token_id="...")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except QuotaExceededError as e:
    print(e.message)
except WatermarkExistsError:
    print("Audio already has a watermark")
except AuthenticationError:
    print("Invalid API key")
except OfSpectrumError as e:
    print(f"API error: {e.code} - {e.message}")

Context Manager

with OfSpectrum(api_key="your_api_key") as client:
    tokens = client.tokens.list()