Skip to content
GitHub

API Reference

Classes

Section titled “Classes”

Talks to a cogvideox_server.py reference server via VideoProvider.
__init__
def __init__(
    base_url: str,
    timeout: float = 180.0,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, VideoGenerationError]

Submits a filled-in SVD workflow to ComfyUI, polls history, fetches output.

Requires request.image_uri — the bundled default workflow targets SVD-style image-to-video graphs (design spec §11.4). image_uri is assumed to already be a path reachable by the ComfyUI process, the same locally-reachable trust boundary the design assumes for base_url itself (design spec §3 non-goals).

__init__
def __init__(
    base_url: str,
    checkpoint: str,
    workflow_path: str | None = None,
    fps: int = 6,
    motion_bucket_id: int = 127,
    poll_interval: float = 1.0,
    timeout: float = 120.0,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, VideoGenerationError]

Runs ffmpeg subprocesses to fulfill VideoOperation requests.
__init__
def __init__(config: VideoProcessingConfig) -> None
process
async def process(
    operation: VideoOperation,
    *,
    progress_callback: Callable[[float], None] | None = None
) -> Result[MediaAsset, VideoProcessingError]
extract_frames
async def extract_frames(
    asset: MediaAsset,
    *,
    fps: float | None = None
) -> Result[list[MediaAsset], VideoProcessingError]
assemble_frames
async def assemble_frames(
    frames: list[MediaAsset],
    *,
    fps: float
) -> Result[MediaAsset, VideoProcessingError]

Talks to a self-hosted video generation HTTP server.

No SDK dependency — this is the always-available zero-extra backend. Video servers commonly respond with either the raw media bytes or a JSON body of the form {"url": "..."} (files are too large to ship inline) — both response shapes are supported.

__init__
def __init__(
    base_url: str,
    timeout: float = 60.0,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, VideoGenerationError]

Talks to an OpenAI-compatible video-generation gateway.
__init__
def __init__(
    api_key: str,
    model: str = 'sora-2',
    base_url: str = 'https://api.openai.com',
    timeout: float = 60.0,
    poll_interval: float = 3.0,
    max_polls: int = 60,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, VideoGenerationError]

Talks to Runway's ML API for video generation.

No SDK dependency — plain HTTP against Runway’s documented async-job API shape: submission returns a job id, a follow-up poll returns a status plus the output URL when the job is done. generate() therefore performs an internal submit + poll loop (bounded retries with a sleep, both configurable) instead of the single round trip the local-http backends perform — it is intentionally slower.

__init__
def __init__(
    api_key: str,
    timeout: float = 60.0,
    poll_interval: float = 3.0,
    max_polls: int = 60,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, VideoGenerationError]

Talks to an svd_server.py reference server via VideoProvider.
__init__
def __init__(
    base_url: str,
    timeout: float = 120.0,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, VideoGenerationError]

Configuration for the video generation subsystem.

Provider that registers a configured VideoProvider backend.
__init__
def __init__(config: VideoConfig | None = None) -> None
register
async def register(container: ContainerRegistrarProtocol) -> None
boot
async def boot(container: ContainerResolverProtocol) -> None

No async I/O needed at boot beyond what register() already did.

health_check
async def health_check(timeout: float = 5.0) -> HealthCheckResult

Callable task handler — wraps a VideoProvider backend for lexigram-tasks.

Returns a plain dict (never raw MediaAsset bytes) because lexigram-tasks’ result store JSON-serializes JobResult. The umbrella wraps this handler during MultimediaProvider.register() to persist any bytes to lexigram-storage BEFORE this dict is constructed — see the “Async job model” section of the design spec.

__init__
def __init__(backend: VideoProvider) -> None
run
async def run(params: dict[str, Any]) -> dict[str, Any]

Video generation integration.
configure
def configure(
    cls,
    config: VideoConfig | None = None
) -> DynamicModule
stub
def stub(
    cls,
    config: VideoConfig | None = None
) -> DynamicModule

Configuration for the local FFmpeg video-processing pipeline.

Task handler for the async video processing job path.

Reconstructs a VideoOperation variant from a flat params dict using the operation_type discriminator (the dataclass’s class name), matching the pattern Timeline.from_params() uses for timeline renders.

__init__
def __init__(backend: VideoProcessor) -> None
run
async def run(params: dict[str, Any]) -> dict[str, Any]

Talks to a wan22_server.py reference server via VideoProvider.
__init__
def __init__(
    base_url: str,
    timeout: float = 180.0,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, VideoGenerationError]

Exceptions

Section titled “Exceptions”

Raised when the video backend rejects the configured API credentials.

Base for video generation errors.

Raised when a video generation operation exceeds its timeout.