lpgcast.serve

lpgcast serve: casting over HTTP, for a platform that holds the schema.

One stateless, synchronous route: POST /cast takes a schema inline, the episodes and options, and returns what lpgcast cast --out writes — the graph, groundings, pending edges, quarantine, episode results and the run manifest. The service never writes to a database and never calls back to the caller, so it holds no database credentials and needs no URL allow-list.

The key is fail-closed: every route but /health needs Authorization: Bearer <LPGCAST_SERVE_API_KEY>, and without a configured key the service refuses to cast unless LPGCAST_SERVE_ALLOW_ANONYMOUS is set. Model settings come from the service's own LPGCAST_LLM_* environment; a request cannot pick the model or the endpoint.

Needs the server extra (FastAPI and uvicorn).

Attributes

Classes

CastOptions

Bases: BaseModel

Per-request settings; each one left out, or null, takes the service's value.

Source code in lpgcast/serve.py

class CastOptions(BaseModel):
    """Per-request settings; each one left out, or null, takes the service's value."""

    model_config = ConfigDict(extra="forbid")

    user_instruction: str | None = None
    slice_max_tokens: int | None = Field(default=None, ge=1)
    slice_max_elements: int | None = Field(default=None, ge=1)
    max_episode_tokens: int | None = Field(default=None, ge=1)

Attributes

max_episode_tokens = Field(default=None, ge=1) class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

slice_max_elements = Field(default=None, ge=1) class-attribute instance-attribute

slice_max_tokens = Field(default=None, ge=1) class-attribute instance-attribute

user_instruction = None class-attribute instance-attribute

CastRequest

Bases: BaseModel

A schema (or a manifest carrying one) and the episodes to cast against it.

Source code in lpgcast/serve.py

class CastRequest(BaseModel):
    """A schema (or a manifest carrying one) and the episodes to cast against it."""

    model_config = ConfigDict(extra="forbid", populate_by_name=True)

    schema_: dict[str, Any] = Field(alias="schema")
    episodes: list[Episode] = Field(min_length=1)
    options: CastOptions = Field(default_factory=CastOptions)

Attributes

episodes = Field(min_length=1) class-attribute instance-attribute

model_config = ConfigDict(extra='forbid', populate_by_name=True) class-attribute instance-attribute

options = Field(default_factory=CastOptions) class-attribute instance-attribute

schema_ = Field(alias='schema') class-attribute instance-attribute

CastResponse

Bases: BaseModel

What lpgcast cast --out writes, in one body.

Source code in lpgcast/serve.py

class CastResponse(BaseModel):
    """What ``lpgcast cast --out`` writes, in one body."""

    graph: dict[str, Any] = Field(
        description="graflo ``GraFloOutput``: schema and data."
    )
    groundings: list[Grounding]
    pending: list[PendingEdge]
    quarantine: list[QuarantineRecord]
    episodes: list[EpisodeResult]
    manifest: RunManifest

Attributes

episodes instance-attribute

graph = Field(description='graflo ``GraFloOutput``: schema and data.') class-attribute instance-attribute

groundings instance-attribute

manifest instance-attribute

pending instance-attribute

quarantine instance-attribute

Health

Bases: BaseModel

GET /health.

Source code in lpgcast/serve.py

class Health(BaseModel):
    """``GET /health``."""

    status: str = "ok"
    lpgcast_version: str
    graflo_version: str
    auth_configured: bool
    max_episodes: int

Attributes

auth_configured instance-attribute

graflo_version instance-attribute

lpgcast_version instance-attribute

max_episodes instance-attribute

status = 'ok' class-attribute instance-attribute

Functions:

create_app(settings=None, serve=None, *, client=None)

The service.

Parameters:

Name	Type	Description	Default
settings	LPGCastSettings | None	Cast settings; defaults to the environment.	None
serve	ServeSettings | None	Service settings; defaults to the environment.	None
client	LLMClient | None	The model client shared by every request; defaults to one built from settings.	None

Source code in lpgcast/serve.py

def create_app(
    settings: LPGCastSettings | None = None,
    serve: ServeSettings | None = None,
    *,
    client: LLMClient | None = None,
) -> FastAPI:
    """The service.

    Args:
        settings: Cast settings; defaults to the environment.
        serve: Service settings; defaults to the environment.
        client: The model client shared by every request; defaults to one
            built from *settings*.
    """
    settings = settings or LPGCastSettings()
    serve = serve or ServeSettings()
    shared = client or LLMClient(settings.llm, LLMCache(settings.cache))
    app = FastAPI(title="LPGCast", version=__version__)

    def require_key(
        authorization: Annotated[str | None, Header()] = None,
    ) -> None:
        expected = serve.api_key.get_secret_value() if serve.api_key else ""
        if not expected:
            if serve.allow_anonymous:
                return
            raise HTTPException(
                503,
                "LPGCAST_SERVE_API_KEY is not configured. Set it, or set "
                "LPGCAST_SERVE_ALLOW_ANONYMOUS=true for local development.",
            )
        given = authorization or ""
        if not hmac.compare_digest(given.encode(), f"Bearer {expected}".encode()):
            raise HTTPException(401, "Invalid or missing bearer token")

    @app.get("/health")
    def health() -> Health:
        return Health(
            lpgcast_version=__version__,
            graflo_version=version("graflo"),
            auth_configured=serve.api_key is not None,
            max_episodes=serve.max_episodes,
        )

    @app.post("/cast", dependencies=[Depends(require_key)])
    async def cast(request: CastRequest) -> CastResponse:
        if len(request.episodes) > serve.max_episodes:
            raise HTTPException(
                413,
                f"{len(request.episodes)} episodes; at most {serve.max_episodes} "
                "per request.",
            )
        try:
            schema = schema_from_mapping(request.schema_)
        except (TypeError, ValueError) as error:
            raise HTTPException(422, f"Not a schema: {error}") from error
        overrides = request.options.model_dump(exclude_none=True)
        run_settings = settings.model_copy(update=overrides)
        sink = MemoryQuarantineSink()
        caster = LPGCaster(schema, settings=run_settings, client=shared, sink=sink)
        try:
            result = await caster.cast(request.episodes)
        except LLMConfigurationError as error:
            raise HTTPException(503, f"The model is misconfigured: {error}") from error
        except CacheMissError as error:
            raise HTTPException(409, str(error)) from error
        return CastResponse(
            graph=result.output.to_dict(),
            groundings=result.groundings,
            pending=result.pending,
            quarantine=sink.records,
            episodes=result.episodes,
            manifest=result.manifest,
        )

    return app

run(serve=None)

Serve on LPGCAST_SERVE_HOST:LPGCAST_SERVE_PORT.

Source code in lpgcast/serve.py

def run(serve: ServeSettings | None = None) -> None:
    """Serve on ``LPGCAST_SERVE_HOST``:``LPGCAST_SERVE_PORT``."""
    import uvicorn

    serve = serve or ServeSettings()
    uvicorn.run(create_app(serve=serve), host=serve.host, port=serve.port)