ContentFlow Python SDK

Podcast intelligence API — transcripts, semantic search, entity extraction, topic classification, and AI summaries from 50,000+ hours of business podcasts.

Installation

pip install contentflow-sdk

Quick Start

from contentflow import ContentFlow

client = ContentFlow(api_key="pk_live_...")

# Search 50,000+ hours of podcast content
results = client.search.hybrid(query="AI fundraising strategies")
for result in results.results:
    print(f"{result.title}: {result.best_chunk_text[:100]}")

# Transcribe any YouTube video
job = client.jobs.create(youtube_url="https://youtube.com/watch?v=...")
job = client.jobs.wait(job.id)  # polls until complete

summary = client.jobs.get_summary(job.id)
print(summary.summaries[0].content)

Async Support

from contentflow import AsyncContentFlow

async with AsyncContentFlow() as client:
    job = await client.jobs.create(youtube_url="https://youtube.com/watch?v=...")
    job = await client.jobs.wait(job.id)
    transcript = await client.jobs.get_transcript(job.id)

Features

• Typed responses — full Pydantic models with IDE autocomplete
• Sync + async — ContentFlow and AsyncContentFlow clients
• Smart polling — jobs.wait() with exponential backoff
• Automatic retries — configurable retry with backoff on transient errors
• Rate limit handling — parsed rate limit headers on RateLimitError
• CLI tool — pip install contentflow-sdk[cli] for terminal access

API Reference

Jobs

Search

Search filters: limit, alpha (keyword vs semantic balance), speaker_ids, entity_ids, topic_ids, channel_ids, date_from, date_to

CLI

pip install contentflow-sdk[cli]
export CONTENTFLOW_API_KEY=pk_live_...

# Submit and wait for transcription
contentflow jobs create "https://youtube.com/watch?v=..." --wait

# Download transcript
contentflow jobs transcript JOB_ID --format markdown --output transcript.md

# Get AI summary
contentflow jobs summary JOB_ID

# Search across all content
contentflow search hybrid "startup fundraising" --limit 5
contentflow search semantic "how to handle investor objections"

Configuration

client = ContentFlow(
    api_key="pk_live_...",            # or set CONTENTFLOW_API_KEY env var
    base_url="https://custom.url",    # default: https://api.gocontentflow.com
    timeout=60.0,                     # request timeout in seconds (default: 30)
    max_retries=5,                    # retry count for transient errors (default: 3)
)

Error Handling

from contentflow import ContentFlow, RateLimitError, NotFoundError, AuthenticationError

client = ContentFlow(api_key="pk_live_...")

try:
    job = client.jobs.get("nonexistent-id")
except NotFoundError:
    print("Job not found")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
    print(f"Remaining: {e.remaining}/{e.limit}")
except AuthenticationError:
    print("Invalid API key")

Exception hierarchy:

ContentFlowError
├── APIError
│   ├── APIConnectionError
│   │   └── APITimeoutError
│   └── APIStatusError
│       ├── BadRequestError        (400)
│       ├── AuthenticationError    (401)
│       ├── PermissionDeniedError  (403)
│       ├── NotFoundError          (404)
│       ├── RateLimitError         (429)
│       └── InternalServerError    (5xx)
├── JobFailedError
└── JobTimeoutError

Get Your API Key

Sign up at gocontentflow.com to get your API key.