ContentFlow
com.gocontentflow/mcp
Documentation
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 —
ContentFlowandAsyncContentFlowclients - 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
| Method | Description |
|---|---|
client.jobs.create(youtube_url=...) | Submit a video for transcription |
client.jobs.get(job_id) | Get job status and details |
client.jobs.wait(job_id) | Poll until job completes |
client.jobs.create_and_wait(youtube_url=...) | Submit and wait in one call |
client.jobs.get_transcript(job_id, format=...) | Download transcript (json/srt/markdown/plain) |
client.jobs.get_summary(job_id) | AI summary with citations |
client.jobs.get_speakers(job_id) | Identified speakers and roles |
client.jobs.get_segments(job_id) | Topical video sections |
client.jobs.get_entities(job_id) | Companies, people, products mentioned |
client.jobs.get_topics(job_id) | Topic classifications |
Search
| Method | Description |
|---|---|
client.search.hybrid(query, ...) | Keyword + semantic search with filters |
client.search.semantic(query, ...) | Pure meaning-based 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.
