Serve¶

The anyinfer.serve frontend: an OpenAI-compatible loopback service over any configured provider, embeddable as an ASGI app. Guide: serve.

Application¶

anyinfer.serve.create_app ¶

create_app(
    client: Any,
    *,
    auth_token: str | None = None,
    expose_targets: Sequence[str] = (),
) -> Any

Build the ASGI application.

Parameters:

Name	Type	Description	Default
`client`	`Any`	An `AsyncClient` to federate through.	required
`auth_token`	`str \| None`	Bearer token clients must present. `None` disables authentication, which is only appropriate on loopback.	`None`
`expose_targets`	`Sequence[str]`	Concrete `provider:model` targets to advertise from `/v1/models`, in addition to catalog aliases.	`()`

Returns:

Type	Description
`Any`	A Starlette application.

Raises:

Type	Description
`ConfigError`	If the `[serve]` extra is not installed.

OpenAI codec¶

The translation layer between the OpenAI wire dialect and AnyInfer's native types (see the architecture overview). Useful directly when embedding the frontend or building a custom edge.

anyinfer.serve.request_from_openai ¶

request_from_openai(
    body: Mapping[str, Any],
) -> tuple[str, GenerationRequest, bool]

Decode an OpenAI chat-completions request body.

Parameters:

Name	Type	Description	Default
`body`	`Mapping[str, Any]`	The parsed request JSON.	required

Returns:

Type	Description
`str`	A `(target, request, stream)` triple. `target` is the `model` field taken
`GenerationRequest`	verbatim — an AnyInfer target is an OpenAI model string (invariant 3), which is
`bool`	what makes federation free.

anyinfer.serve.request_to_openai ¶

request_to_openai(
    target: str,
    request: GenerationRequest,
    *,
    stream: bool = False,
) -> dict[str, Any]

Encode a request back into OpenAI wire form.

The inverse of request_from_openai(), and the basis of the round-trip test that enforces invariant 1: anything the OpenAI surface can express must survive the trip.

anyinfer.serve.completion_from_generation ¶

completion_from_generation(
    result: Generation,
    *,
    model: str,
    completion_id: str = "chatcmpl-anyinfer",
    created: int | None = None,
) -> dict[str, Any]

Render a Generation as a chat.completion object.

anyinfer.serve.chunk_from_event ¶

chunk_from_event(
    event: StreamEvent,
    *,
    model: str,
    completion_id: str = "chatcmpl-anyinfer",
    created: int | None = None,
) -> dict[str, Any] | None

Render one stream event as a chat.completion.chunk.

Returns None for events with no OpenAI equivalent (timing marks, attempt records) — they are AnyInfer-native observability that the OpenAI wire format cannot carry.

anyinfer.serve.final_chunk ¶

final_chunk(
    result: Generation,
    *,
    model: str,
    completion_id: str = "chatcmpl-anyinfer",
    created: int | None = None,
    include_usage: bool = True,
) -> Iterable[dict[str, Any]]

Render the terminal chunks: the finish reason, then optionally usage.

Usage rides in its own trailing chunk with an empty choices array, matching stream_options.include_usage. Clients that stop reading at finish_reason miss it — which is exactly the bug the core's own parser is written to avoid.

anyinfer.serve.encode_messages ¶

encode_messages(
    messages: Sequence[Message],
) -> list[dict[str, Any]]

Encode typed messages back into an OpenAI messages array.

anyinfer.serve.decode_messages ¶

decode_messages(
    raw: Sequence[Mapping[str, Any]],
) -> tuple[Message, ...]

Decode an OpenAI messages array into typed messages.