Capture SDK Overview

import videodb

conn = videodb.connect()

# Create session for a user
cap = conn.create_capture_session(
    end_user_id="user_abc",
    callback_url="https://your-backend.com/webhooks/videodb",
    metadata={"app": "my-ai-copilot"}
)

# Generate token for desktop client (never share API key)
token = conn.generate_client_token(expires_in=600)

# Send session ID and token to desktop client
print(f"Session: {cap.id}, Token: {token}")

import asyncio
from videodb.capture import CaptureClient

async def capture(capture_session_id: str, client_token: str):
    client = CaptureClient(client_token=client_token)

    # Request permissions
    await client.request_permission("microphone")
    await client.request_permission("screen_capture")

    # Discover available sources
    channels = await client.list_channels()
    mic = channels.mics.default
    display = channels.displays.primary or channels.displays[1]
    system_audio = channels.system_audio.default
    selected = [c for c in [mic, display, system_audio] if c]

    # Start capture
    await client.start_session(
        capture_session_id=capture_session_id,
        channels=selected,
        primary_video_channel_id=display.name if display else None
    )

    # Listen for events
    async for ev in client.events():
        print(f"{ev.event}: {ev.payload}")
        if ev.event in ("recording-complete", "error"):
            break

    await client.stop_session()
    await client.shutdown()

# Run the capture
if __name__ == "__main__":
    asyncio.run(capture(
        capture_session_id="cap-xxx",  # From backend
        client_token="token-xxx"        # From backend
    ))

def on_webhook(payload: dict):
    if payload["event"] == "capture_session.active":
        cap_id = payload["capture_session_id"]
        cap = conn.get_capture_session(cap_id)

        # Get RTStreams (one per channel)
        mics = cap.get_rtstream("mic")
        displays = cap.get_rtstream("display")

        # Start real-time AI processing
        if mics:
            mic = mics[0]
            mic.start_transcript()
            mic.index_audio(prompt="Extract key decisions and action items")

        if displays:
            display = displays[0]
            display.index_visuals(prompt="Describe what the user is doing")

rtstream.start_transcript()
rtstream.index_audio(prompt="Extract key decisions")
rtstream.index_visuals(prompt="Describe what user is doing")
rtstream.search("budget discussion")