CreativAI Python SDK

Official Python SDK for the CreativAI Video Intelligence Platform.

Upload, index, search, and extract structured knowledge from video libraries at scale — all from Python.

---

Installation

pip install creativai

For SSE streaming support (agentic chat, live stream events), httpx-sse is installed automatically as a dependency.

Requires Python 3.9+

---

Authentication

Get your API key from the CreativAI app (profile avatar → API Key). Keys begin with sk_live_.

import creativai

# Option 1 — pass directly
client = creativai.CreativAI(api_key="sk_live_...")

# Option 2 — environment variable (recommended for production)
# export CREATIVAI_API_KEY="sk_live_..."
client = creativai.CreativAI()

---

MCP Quick Start — Use CreativAI inside Claude, Cursor, and Copilot

CreativAI supports the Model Context Protocol (MCP), letting any compatible AI assistant call CreativAI tools directly in conversation.

Option A — npx (no Python install needed, recommended for most users)

Add to your Claude Desktop / Cursor config:

{
  "mcpServers": {
    "creativai": {
      "command": "npx",
      "args": ["-y", "creativai-mcp"],
      "env": { "CREATIVAI_API_KEY": "sk_live_..." }
    }
  }
}

Option B — pip

pip install creativai-mcp
CREATIVAI_API_KEY="sk_live_..." creativai-mcp          # stdio (Claude Desktop)
creativai-mcp --transport sse --port 8090              # HTTP/SSE server

Option C — from the SDK

import creativai

client = creativai.CreativAI()
server = client.as_mcp_server()   # returns a FastMCP instance
server.run(transport="stdio")     # or transport="sse"

HTTP/SSE (hosted, no install)

Connect directly to the CreativAI backend — no local binary needed:

{
  "mcpServers": {
    "creativai": {
      "type": "sse",
      "url": "https://creativai-apis.com/api/v2/mcp/sse",
      "headers": { "X-API-Key": "sk_live_..." }
    }
  }
}

Available tools (42 total): collections, media, indexing, search, agentic chat, knowledge extraction, data plates, tasks, live stream, online search, YouTube, organizations/projects, account info.

MCP setup page: creativ-ai.com/mcp

---

Quick Start

import time
import creativai

client = creativai.CreativAI()

# Verify your key and check credits
info = client.users.get_users_info()
print(f"Credits: {info['credits']}")

# Create a collection
collection = client.collections.create("my-dashcam-footage", model="video_only")
cid = collection["collection_id"]

# Upload a local file
client.media.upload_file(cid, "dashcam_2026.mp4")

# Start indexing (async — returns immediately with a job ID)
job = client.indexing.start(cid)
indexing_id = job["indexing_id"]

# Poll until complete
while True:
    status = client.indexing.get_status(indexing_id)
    if status["status"] == "completed":
        break
    time.sleep(10)

# Semantic search
results = client.search.query(cid, "pedestrian crossing the road")
for hit in results["results"][:5]:
    print(f"[{hit['score']:.2f}] {hit['video_name']} @ {hit['start_time']}s")

---

Resource Reference

All resources are accessed as attributes on the CreativAI client instance.

client.health

client.health.check()        # GET /health
client.health.versioned()    # GET /api/v2/health

client.users

client.users.me()
client.users.info()
client.users.get_users_info()
client.users.claim_welcome_credits()

client.collections

client.collections.create("name", model="video_only")   # model: "video_only" | "multimodal"
client.collections.list()
client.collections.get(collection_id)
client.collections.update(collection_id, collection_name="new-name")
client.collections.delete(collection_id)
client.collections.restore(collection_id)
client.collections.list_by_organization(org_id)
client.collections.list_by_project(org_id, project_name)

client.media

client.media.list(collection_id)
client.media.upload_file(collection_id, "/path/to/video.mp4")  # convenience helper
client.media.get_upload_url(collection_id, "video.mp4")        # get presigned URL
client.media.get_upload_urls(collection_id, ["a.mp4", "b.mp4"])
client.media.delete(collection_id, ["s3://bucket/key1.mp4"])

client.uploads — multipart

upload = client.uploads.initiate(collection_id, "large-video.mp4")
client.uploads.complete(upload["upload_id"], parts=[{"part_number": 1, "etag": "..."}])
client.uploads.abort(upload["upload_id"])
client.uploads.regenerate_urls(upload["upload_id"])

client.transfers — external S3 / URL

job = client.transfers.start(collection_id, "s3://my-bucket/video.mp4")
client.transfers.get_status(job["job_id"])
client.transfers.validate("https://example.com/video.mp4")

client.indexing

job = client.indexing.start(collection_id)
client.indexing.get_status(job["indexing_id"])
client.indexing.estimate_cost(collection_id)
client.indexing.get_preprocessing_status(collection_id)
client.indexing.list_preprocessed_videos(collection_id)

client.search

results = client.search.query(
    collection_id,
    "person wearing PPE",
    search_type="hybrid",   # "hybrid" | "vision" | "audio"
    page_number=1,
    page_size=50,
    refine_query=True,
)

client.data_plates

plate_job = client.data_plates.create_from_collection(collection_id, plate_name="All Segments")
plate_id = poll_until_done(client.data_plates.get_creation_job, plate_job["job_id"])["plate_id"]

plate = client.data_plates.get(collection_id, plate_id, page_size=100)
client.data_plates.update(collection_id, plate_id, plate_name="Renamed")
client.data_plates.delete(collection_id, plate_id)

# Segments
client.data_plates.add_segments(collection_id, plate_id, segments=[...])
client.data_plates.remove_segments(collection_id, plate_id, segment_ids=["seg_1"])
client.data_plates.update_extracted_info(collection_id, plate_id, "seg_1", "ppe_worn", True)

# Export
client.data_plates.generate_csv(collection_id, plate_id)
csv_bytes = client.data_plates.export_csv(collection_id, plate_id)

client.knowledge_extraction

ke_job = client.knowledge_extraction.add_columns(
    collection_id,
    plate_id,
    columns=[
        {"name": "ppe_worn", "question": "Is PPE worn?", "type": "boolean"},
        {"name": "activity",  "question": "What is happening?", "type": "text"},
    ],
)
client.knowledge_extraction.get_job(ke_job["job_id"])

# AI chat query over the plate data
answer = client.knowledge_extraction.chat_query(collection_id, plate_id, "How many PPE violations?")
print(answer["answer"])

# Charts
charts = client.knowledge_extraction.get_plate_charts(collection_id, plate_id)

client.agentic_chat — SSE streaming

session = client.agentic_chat.create_session(collection_id, title="My analysis")
sid = session["session_id"]

for event in client.agentic_chat.chat(sid, "Find all forklift incidents and summarize them"):
    match event["event"]:
        case "thinking":
            print(f"  [thinking] {event['data'].get('text', '')[:80]}")
        case "search":
            print(f"  [search] {event['data']}")
        case "answer":
            print(f"\n{event['data'].get('text', '')}")
        case "done":
            break

# Session management
client.agentic_chat.list_sessions(collection_id=collection_id)
client.agentic_chat.get_messages(sid)
client.agentic_chat.stop(sid)
client.agentic_chat.delete_session(sid)

client.live_stream

# RTMP push — point OBS or ffmpeg at publish_url
session = client.live_stream.stream_rtmp(
    collection_id=collection_id,
    name="Entrance Camera",
    model="video_only",
)
print(session["publish_url"])

# RTSP pull — IP camera
session = client.live_stream.stream_rtsp("rtsp://192.168.1.100/stream", collection_id=collection_id)

# WebRTC — browser webcam
session = client.live_stream.stream_webrtc(collection_id=collection_id)
print(session["whip_url"], session["whep_url"])

# Add questions and poll
client.live_stream.add_questions(sid, ["Is anyone present?", "Is the door open?"])
client.live_stream.stop_session(sid)

client.upload_integrations

# Google Drive
files = client.upload_integrations.google_drive_list_files(google_access_token)
client.upload_integrations.google_drive_transfer(
    collection_id, google_access_token,
    file_ids=["drive_file_id"], file_names=["video.mp4"]
)

# Dropbox
files = client.upload_integrations.dropbox_list_files(dropbox_access_token)
client.upload_integrations.dropbox_transfer(
    collection_id, dropbox_access_token,
    file_paths=["/Videos/clip.mp4"], file_names=["clip.mp4"]
)

# Hugging Face
files = client.upload_integrations.huggingface_list_files(hf_token, "username/my-dataset")
client.upload_integrations.huggingface_transfer(
    collection_id, hf_token, "username/my-dataset",
    file_paths=["videos/clip.mp4"]
)

client.organizations / client.projects

org = client.organizations.create("Acme Corp")
client.projects.create(org["org_id"], "production-analysis")
client.projects.list(org["org_id"])

client.sharing

client.sharing.invite(collection_id, "alice@example.com", role="viewer")
client.sharing.list_members(collection_id)
client.sharing.update_member(collection_id, user_id, role="editor")
client.sharing.remove_member(collection_id, user_id)
client.sharing.create_group(collection_id, "annotators")

client.tasks

task = client.tasks.create(collection_id, title="Review batch 1", assigned_to=[user_id])
client.tasks.update_status(task["task_id"], "in_progress")
client.tasks.update_progress(task["task_id"], 50)
client.tasks.add_comment(task["task_id"], "Segment 12 flagged for review")
client.tasks.my_tasks()

client.transactions / client.subscriptions / client.invoices

client.transactions.summary()
client.transactions.breakdown_by_collections()
client.transactions.export()  # CSV bytes

client.subscriptions.current()
client.subscriptions.list_plans()

client.invoices.list()
pdf = client.invoices.download("inv_123")

client.jobs — cancel any async job

client.jobs.cancel("indexing-chunk", "idx_abc123")
client.jobs.cancel("knowledge-extraction", "ke_job_xyz")

---

Error Handling

import creativai

client = creativai.CreativAI()

try:
    results = client.search.query("col_invalid", "query")
except creativai.NotFoundError as e:
    print(f"Not found: {e.message}")
except creativai.InsufficientCreditsError:
    print("Top up your credits at https://creativ-ai.com/pricing")
except creativai.AuthenticationError:
    print("Check your CREATIVAI_API_KEY")
except creativai.APIError as e:
    print(f"API error {e.status_code}: {e.message} (code={e.code})")

Exception	HTTP status
AuthenticationError	401
InsufficientCreditsError	402
PermissionError	403
NotFoundError	404
ValidationError	400 / 422
RateLimitError	429
ServerError	5xx
StreamingError	SSE connection failure
TimeoutError	Request timeout

---

Context Manager

with creativai.CreativAI() as client:
    collections = client.collections.list()
# HTTP connection pool closed automatically

---

Examples

File	Description
examples/quickstart.py	Upload, index, search, and agentic chat
examples/knowledge_extraction.py	Structured data extraction → CSV
examples/live_stream.py	RTMP live stream session

---

License

MIT