pip install videorouter-live

videorouter-live>=0.1

from videorouter_live import LiveVideo

lv = LiveVideo(api_key="llmr_sk_live_...")     # or set VIDEOROUTER_API_KEY

with lv.session(
    "fal/happy-oyster-adventure",
    max_duration_seconds=60,
    options={"world": {"prompt": "a rainy cyberpunk city", "perspective": "first_person"}},
) as s:
    # The session is already ready and heartbeating in the background.
    send_to_browser(s.connect)         # host credentials - returned once by the server
    wait_for_user_to_finish()

# Settled on exit, even if your code raised.
print(s.status, s.billed_seconds, s.cost_usd, s.end_reason)

import asyncio
from videorouter_live import AsyncLiveVideo

async def main():
    async with AsyncLiveVideo(api_key="llmr_sk_live_...") as lv:
        async with lv.session("decart/lucy-2.5", max_duration_seconds=120,
                              options={"speed": "standard"}) as s:
            s.on("error", lambda e: print("heartbeat problem:", e))
            await hand_to_browser(s.connect)
            await asyncio.sleep(30)
        print(s.cost_usd)

asyncio.run(main())

s = lv.create_session("fal/happy-oyster-adventure", max_duration_seconds=60,
                      options={"world": {"prompt": "x", "perspective": "third_person"}})
print(s.status)                       # "waiting" - a world is building, nothing is metered
s.on("status_changed", lambda st: print("->", st))
s.on("ended", lambda sess: print("settled:", sess.cost_usd))

s.wait_until_ready(timeout=180, poll_interval=4)   # polls; the poll that sees it ready returns s.connect
s.start_heartbeat()                   # daemon thread, every ~5 s
use(s.connect)

s.end(host_travel_id="...")           # optional (Happy Oyster): bill on the host's own duration and fetch the replay

token = lv.create_token(models=["fal/happy-oyster-adventure"], expires_in=300,
                        max_sessions=1, max_session_duration_seconds=60)
token.token        # "llmr_sk_live_..." - give this to the browser
token.expires_at   # ISO timestamp

from videorouter_live import LiveVideo

lv = LiveVideo()                           # reads VIDEOROUTER_API_KEY

# Create and wait
job = lv.videos.generate({
    "model": "heygen-video-1",
    "prompt": "A presenter says: 'Welcome to VideoRouter.'",
    "duration_secs": 5,
})
print(job.url)

# Or step by step
queued = lv.videos.create({"model": "wan-3.0-prime", "prompt": "A robotic arm in a lab"})
queued.wait(timeout=600, poll_interval=5)  # raises VideoGenerationError / TimeoutError
again = lv.videos.get(queued.id)

from videorouter_live import (InsufficientCreditsError, ConcurrencyLimitError, HostCapacityError,
                              PricingUnavailableError, LiveVideoError)

try:
    s = lv.create_session("fal/happy-oyster-adventure", max_duration_seconds=60)
except InsufficientCreditsError:
    ...                                   # top up, or book fewer seconds
except ConcurrencyLimitError as e:
    ...                                   # end a session first; e.retry_after is set
except HostCapacityError as e:
    time.sleep(e.retry_after or 5)        # the host is full right now - sessions are never queued
except PricingUnavailableError:
    ...                                   # model.available is False
except LiveVideoError as e:
    print(e.status, e.code, e.message)

import httpx
from videorouter_live import LiveVideo

def handler(request: httpx.Request) -> httpx.Response:
    return httpx.Response(200, json={"object": "list", "data": []})

lv = LiveVideo(api_key="test", base_url="http://test/v1",
               http_client=httpx.Client(transport=httpx.MockTransport(handler)))
assert lv.models() == []