Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.sf-voice.sh/llms.txt

Use this file to discover all available pages before exploring further.

sf-voice-media — built on httpx. Sync and async clients with identical interfaces. Use the sync client for scripts and workers. Use the async client inside FastAPI, Starlette, or other async services.

Install

pip install sf-voice-media==0.1.1
Python 3.9+. Dependency: httpx >= 0.27.

Sync client

from sf_voice import SfVoiceMedia

client = SfVoiceMedia(
    api_key="your_api_key",
    base_url="https://api.sf-voice.com",
)

# use as a context manager to close the connection pool automatically
with SfVoiceMedia(api_key="...") as client:
    ...

Async client

from sf_voice import AsyncSfVoiceMedia
import asyncio

async def main():
    async with AsyncSfVoiceMedia(api_key="...") as client:
        ...

asyncio.run(main())
The async client has the same method names. Each method is a coroutine.

Ingest

# from a URL
resp = client.ingest(
    source="url",
    asset_id="call_001",
    asset_class="customer_acme",
    url="https://storage.example.com/calls/001.mp3",
    media_type="audio",
    types=["audio", "transcript"],
    metadata={"caller": "+14155550199"},
)
print(resp.task_id)

# from a file
with open("call.mp3", "rb") as f:
    resp = client.ingest(
        source="file",
        asset_id="call_001",
        asset_class="customer_acme",
        file=f.read(),
        filename="call.mp3",
        content_type="audio/mpeg",
        media_type="audio",
        types=["audio", "transcript"],
    )

Poll until ready

task = client.poll_task(
    resp.task_id,
    interval_s=2.0,
    timeout_s=120.0,
)

if task.status == "failed":
    raise RuntimeError(task.error or "indexing failed")
poll_task raises SfVoiceMediaPollTimeoutError if the task doesn’t complete within timeout_s.
result = client.search(
    query="customer asks about pricing",
    asset_class="customer_acme",
    types=["transcript"],
    threshold=0.7,
    limit=10,
)

for r in result.results:
    print(r.asset_id, r.start_ms, r.end_ms, r.score)
Search by specific assets:
result = client.search(
    query="cancellation request",
    asset_ids=["call_001", "call_002"],
)

Assets

# list
resp = client.list_assets(page=1, limit=20)
for asset in resp.items:
    print(asset.asset_id, asset.status)

# get one
asset = client.get_asset("call_001")

# delete
client.delete_asset("call_001")

Errors

from sf_voice import SfVoiceMedia, SfVoiceMediaError, SfVoiceMediaPollTimeoutError

try:
    task = client.poll_task(task_id, timeout_s=60.0)
except SfVoiceMediaPollTimeoutError as e:
    print(f"timed out waiting for {e.task_id}")
except SfVoiceMediaError as e:
    print(e.code, e.status, e.message)
ExceptionWhen
SfVoiceMediaErrorNon-2xx API response. Carries code, message, status.
SfVoiceMediaRequestTimeoutErrorHTTP request timed out. Carries timeout_ms.
SfVoiceMediaPollTimeoutErrorpoll_task timed out. Carries task_id, timeout_ms.