Anchorcast

A Python client for real-time media streams generated by pluggable models.

Early (0.1.0). The session / model / source interface is the point. Expect it to move.

Install

pip install anchorcast

From git:

pip install git+https://github.com/scrollmark/anchorcast.git

For local development:

git clone https://github.com/scrollmark/anchorcast.git
cd anchorcast
python3.11 -m pip install -e ".[dev]"
python3.11 -m pytest

Requirements: Python 3.11+, ffmpeg on PATH (for last-frame continuity), and a fal key if you use the H3 Max adapter (FAL_KEY or FALAI_API_KEY).

Quickstart

from pathlib import Path

from anchorcast import Brief, Character, QueueSource, Session
from anchorcast.models import H3Max

source = QueueSource()
session = Session(
    model=H3Max(),
    character=Character(
        name="Bear",
        image=Path("bear.png"),
        visual_prompt="Hand-drawn stuffed bear on cream paper.",
        voice_prompt="warm, unhurried cadence",
    ),
    source=source,
    workdir=Path(".anchorcast"),
)

source.submit(
    Brief(
        id="rates",
        topic="mortgage rates",
        talking_points=("Why are mortgage rates still so high?",),
        citations=("https://example.com/rates",),
        priority=10,
    )
)

for event in session.run(max_segments=1):
    print(event.segment.path, event.segment.brief.topic)

A Brief is the next beat of the stream. The session asks the source for one, asks the model for a clip, and keeps a small generate-ahead buffer so you do not mint faster than the playhead.

Core types

Sources

• QueueSource — explicit briefs, ordered by priority
• FeedSource — RSS/Atom mapped to cited stories
• IdleSource — character bits that make no news claims

When the queue is empty, pass idle=IdleSource(character) so the session can keep breathing without inventing facts.

Custom model

from anchorcast.types import GenerateRequest, Segment

class MyModel:
    def generate(self, request: GenerateRequest) -> Segment:
        # write a video to request.output_path
        return Segment(
            path=request.output_path,
            duration=15.0,
            generated_in=1.0,
            brief=request.brief,
        )

Custom source

from anchorcast.types import Brief, Playhead

class WebhookSource:
    def next_brief(self, playhead: Playhead | None = None) -> Brief | None:
        return self._take_from_queue()

YouTube Live, Twitch, TikTok Live, and Super Chats are a later backend. That layer should map platform events into Briefs and submit them to a QueueSource. It does not belong in this client.

Status

• Generate-ahead Session with a pluggable model and source
• fal MiniMax H3 Max adapter (minimax/h3-max/image-to-video)
• Feed, queue, and idle sources
• Last-frame continuity via ffmpeg

Not yet: RTMP output, hosted APIs, or platform chat integrations.

Example app

examples/literary_bear is a continuous 1926 literary-bear news stream (Shepard, not Disney). Each clip is the next fifteen seconds of one monologue, not a fresh "oh bother" cold open.

python3.11 -m examples.literary_bear --dry-run   # scripts only
python3.11 -m examples.literary_bear            # local player at http://127.0.0.1:8765

--llm or OPENAI_API_KEY writes spoken lines with gpt-5.6-luna. --no-llm forces the rotating templates.

Development

python3.11 -m pip install -e ".[dev]"
python3.11 -m pytest

License

MIT. See LICENSE.

The example still under examples/characters/literary-bear/ is an E. H. Shepard illustration from Winnie-the-Pooh (1926), U.S. public domain. It is not Disney, and this project is not affiliated with Disney.