Sidecar Frontend¶

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] = (),
    context_tuning: Any = None,
    max_request_bytes: int = DEFAULT_MAX_REQUEST_BYTES,
) -> 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.	`()`
`context_tuning`	`Any`	Default `ContextTuning` for wire context requests that omit their own `tuning` block — normally the deployment's configured `AnyInferConfig.context`. `None` keeps the library defaults.	`None`
`max_request_bytes`	`int`	Reject a request body larger than this with 413. Every handler buffers its whole body to parse JSON, so without a cap a single client can force unbounded allocation — which matters most when the gateway is exposed off loopback. `0` disables the check.	`DEFAULT_MAX_REQUEST_BYTES`

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], *, context_tuning: Any = None
) -> 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
`context_tuning`	`Any`	Default `ContextTuning` for a request whose context extension omits its own `tuning`. The gateway supplies the deployment's configured tuning, so wire callers inherit it rather than always falling back to the library defaults. Typed opaquely on purpose: the sidecar is a codec and does not import the context implementation, so it forwards the object onward without inspecting it.	`None`

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,
    include_manifest: bool = False,
) -> dict[str, Any]

Render a Generation as a chat.completion object.

Parameters:

Name	Type	Description	Default
`result`	`Generation`	The generation to render.	required
`model`	`str`	The `model` string to echo back.	required
`completion_id`	`str`	The completion id to stamp.	`'chatcmpl-anyinfer'`
`created`	`int \| None`	Unix timestamp; defaults to now.	`None`
`include_manifest`	`bool`	Attach the run manifest under `MANIFEST_FIELD`. Off by default, so a stock client's response is byte-identical to what it was before manifests existed. Serialization only — nothing here assembles a manifest.	`False`

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.