API Reference

Everything exported from primfunctions.completions.

from primfunctions.completions import (
    # Setup
    configure,
    configure_provider,
    close,
    # Top-level calls
    generate_chat_completion,
    generate_chat_completion_stream,
    # Errors
    CompletionsNotConfiguredError,
    CompletionsProviderNotConfiguredError,
    CompletionsProxyError,
    # Request / response
    ChatCompletionRequest,
    ChatCompletionResponse,
    FallbackRequest,
    RetryConfiguration,
    StreamOptions,
    ToolChoice,
    # Tools
    ToolDefinition,
    FunctionDefinition,
    ToolCall,
    FunctionCall,
    # Messages
    AssistantMessage,
    ConversationHistory,
    ConversationHistoryMessage,
    SystemMessage,
    ToolResultMessage,
    UserMessage,
    serialize_conversation,
    deserialize_conversation,
    # Cache
    CacheBreakpoint,
    # Streaming chunks
    ChatCompletionChunk,
    ContentDeltaChunk,
    ContentSentenceChunk,
    ErrorChunk,
    FinalResponseChunk,
    FinishReasonChunk,
    ToolCallChunk,
    UsageChunk,
    # Provider enum
    CompletionsProvider,
)

Setup#

configure_provider#

Register a provider for the current session. Must be called before any generate_chat_completion[_stream] for that provider — including fallback providers.

def configure_provider(
    provider: str | CompletionsProvider,
    *,
    voicerun_managed: bool = False,
    api_key: Optional[str] = None,
    credentials: Optional[dict[str, Any]] = None,
) -> None

• voicerun_managed=True — the proxy uses VoiceRun's mounted key for this provider. Your handler never sees it.
• api_key=<str> — register a customer-supplied single-string key for this session only.
• credentials=<dict> — register a provider-specific credential dict for providers whose auth isn't a single string (e.g. AWS SigV4 for anthropic_bedrock). The dict is injected into the request's provider_kwargs[provider] for you, so credentials never ride on individual completion requests. Combine with api_key when a provider needs both (e.g. openai_bedrock's bearer token + region).

voicerun_managed is mutually exclusive with api_key/credentials, and at least one of the three must be supplied. Otherwise raises ValueError. See AWS Bedrock for the credential shapes.

Not every provider supports voicerun_managed=True. Bring-your-own-credential providers — anthropic_bedrock, google_vertex, openai_bedrock, and groq — have no VoiceRun-mounted key, so they must be registered with api_key/credentials; a voicerun_managed=True registration fails at request time.

The first call for a given provider also kicks off a background warm request to the proxy, paying the TLS handshake during a quiet moment (typically while your greeting is playing).

See Provider Configuration for details.

configure#

Set the proxy URL and auth token. You normally do not call this from a handler — the VoiceRun sandbox runtime configures it for you before your handler code runs.

def configure(proxy_url: str, proxy_token: str) -> None

close#

Close the module-level aiohttp session. Runtimes handle this during shutdown; handler code should not call it.

async def close() -> None

Top-level calls#

generate_chat_completion#

Send a non-streaming completion request. Returns the full ChatCompletionResponse.

async def generate_chat_completion(
    request: ChatCompletionRequest | dict,
) -> ChatCompletionResponse

On transport-level failures it retries once with a fresh connection. Retries configured via RetryConfiguration happen server-side on the proxy.

Raises:

• CompletionsNotConfiguredError — configure(url, token) was never called in this process (the runtime should have done this).
• CompletionsProviderNotConfiguredError — a provider in the request (primary or fallback) was not registered via configure_provider.
• CompletionsProxyError — the proxy returned a non-200 status; the error carries message, error_type, and status_code.
• ValueError — api_key was set on the request body (it must go through configure_provider).

generate_chat_completion_stream#

Send a streaming completion request. Returns an async iterable of ChatCompletionChunk.

async def generate_chat_completion_stream(
    request: ChatCompletionRequest | dict,
    stream_options: StreamOptions | dict | None = None,
) -> AsyncIterable[ChatCompletionChunk]

Connection retries fire before the first chunk only. Mid-stream errors are re-raised as CompletionsProxyError inside async for.

Errors#

class CompletionsNotConfiguredError(Exception): ...
class CompletionsProviderNotConfiguredError(Exception): ...
class CompletionsProxyError(Exception):
    message: str
    error_type: str
    status_code: Optional[int]

Request types#

ChatCompletionRequest#

Main request object.

@dataclass
class ChatCompletionRequest:
    provider: CompletionsProvider | str          # openai | openai_responses | anthropic | google | google_vertex | anthropic_vertex | anthropic_bedrock | openai_bedrock | alibaba | deepseek | groq | voicerun
    model: str                                   # free-form model id, e.g. "gemini-3.5-flash"
    messages: ConversationHistory | list[dict]   # conversation messages
    temperature: Optional[float] = None
    tools: Optional[list[ToolDefinition | dict]] = None
    tool_choice: Optional[ToolChoice] = None
    timeout: Optional[float] = None              # seconds
    max_tokens: Optional[int] = None
    response_schema: Optional[dict[str, Any]] = None  # JSON Schema for structured output
    retry: Optional[RetryConfiguration | dict] = None
    fallbacks: Optional[list[FallbackRequest | dict]] = None
    provider_kwargs: Optional[ProviderKwargs] = None

api_key is not a field. Provider keys come from configure_provider. Attempting to set api_key (or pass it in a dict request) raises ValueError.

provider is a closed set; model is not. An unrecognized provider is rejected by the proxy and surfaces as CompletionsProxyError with error_type == "InvalidProviderError". model is a required free-form string that the proxy forwards to the provider SDK verbatim — there is no model enum, alias table, or rewrite step, so validity is decided by the upstream provider. The one exception: an attempt that would spend a VoiceRun-managed key is checked against the set of models VoiceRun can bill for, and may be rejected before dispatch as CompletionsProxyError with error_type == "ModelNotAllowedError" — supply your own api_key for that provider, or request support for the model. Attempts carrying your own key are exempt, as are google_vertex, anthropic_bedrock, and voicerun. Model ids throughout these docs are examples, not a supported-models list.

FallbackRequest#

Override fields for a fallback provider. Everything unset inherits from the primary.

@dataclass
class FallbackRequest:
    provider: Optional[CompletionsProvider | str] = None
    model: Optional[str] = None
    messages: Optional[ConversationHistory] = None
    temperature: Optional[float] = None
    tools: Optional[list[ToolDefinition | dict]] = None
    tool_choice: Optional[ToolChoice] = None
    timeout: Optional[float] = None
    max_tokens: Optional[int] = None
    response_schema: Optional[dict[str, Any]] = None
    retry: Optional[RetryConfiguration | dict] = None
    provider_kwargs: Optional[ProviderKwargs] = None

api_key is also not a field on fallbacks. The proxy resolves keys from configure_provider per fallback entry.

ProviderKwargs#

Provider-specific kwargs keyed by provider name. Only the entry matching the executing provider is applied.

class ProviderKwargs(TypedDict, total=False):
    openai:            OpenAIKwargs
    openai_responses:  OpenAIResponsesKwargs
    anthropic:         AnthropicKwargs
    google:            GoogleKwargs
    google_vertex:     GoogleVertexKwargs
    anthropic_vertex:  AnthropicVertexKwargs
    anthropic_bedrock: AnthropicBedrockKwargs
    openai_bedrock:    OpenAIBedrockKwargs
    alibaba:           AlibabaKwargs
    deepseek:          DeepSeekKwargs
    groq:              GroqKwargs

OpenAIKwargs

class OpenAIKwargs(TypedDict, total=False):
    service_tier: Literal["auto", "default", "flex", "scale", "priority"]
    reasoning_effort: Literal["none", "minimal", "low", "medium", "high", "xhigh"]

See the OpenAI API reference.

OpenAIResponsesKwargs

Tuning options for the Responses API provider.

class OpenAIResponsesKwargs(TypedDict, total=False):
    reasoning: dict[str, Any]        # e.g. {"effort": "low"}
    include: list[str]               # e.g. ["reasoning.encrypted_content"]
    max_output_tokens: int           # Responses naming for max_tokens
    store: bool                      # persist server-side; defaults to False
    prompt_cache_key: str            # routing hint for automatic prompt caching
    prompt_cache_retention: Literal["in_memory", "24h"]
    service_tier: str                # e.g. "default", "flex", "priority"

Only max_output_tokens and store are consumed by the proxy; every other key — including the ones listed above — is forwarded verbatim as a top-level responses.create parameter, so anything the Responses API accepts can be passed through. max_output_tokens is not just a cap: when omitted it defaults to the request's max_tokens.

See the OpenAI Responses API reference.

AnthropicKwargs

class AnthropicKwargs(TypedDict, total=False):
    thinking: ThinkingConfigParam  # e.g. {"type": "enabled", "budget_tokens": 10000}

See the Anthropic messages API reference.

GoogleKwargs

class GoogleKwargs(TypedDict, total=False):
    thinking_config: ThinkingConfigDict  # e.g. {"thinking_budget": 10000, "include_thoughts": True}

See the Google GenerateContent reference.

GoogleVertexKwargs

Gemini on Google Cloud Vertex AI. Supply the service-account credentials via configure_provider(credentials=...) (recommended) rather than on each request; request-time kwargs hold tuning options like thinking_config, not secrets. See Provider Configuration.

class GoogleVertexKwargs(TypedDict, total=False):
    thinking_config: ThinkingConfigDict
    project_id: str
    location: str
    region: str                                  # alias for location (anthropic_vertex parity)
    service_account_credentials: dict[str, Any]  # parsed SA JSON

project_id, location, and service_account_credentials are all required — a request missing any of them raises CompletionsProxyError with error_type == "ConfigurationError". region is accepted as an alias for location on both paths: request-time provider_kwargs and the configure_provider(credentials=...) registration.

AnthropicVertexKwargs

class AnthropicVertexKwargs(TypedDict, total=False):
    thinking: ThinkingConfigParam
    project_id: str
    region: str
    service_account_credentials: dict[str, Any]  # parsed SA JSON

AnthropicBedrockKwargs

Claude on AWS Bedrock. Authenticates with the caller's own AWS credentials (SigV4) — supply them via configure_provider(credentials=...) (recommended) rather than on each request. See AWS Bedrock.

class AnthropicBedrockKwargs(TypedDict, total=False):
    thinking: ThinkingConfigParam
    aws_access_key: str       # required
    aws_secret_key: str       # required
    aws_region: str           # required
    aws_session_token: str    # optional — for temporary/STS credentials

OpenAIBedrockKwargs

OpenAI models on AWS Bedrock via its OpenAI-compatible endpoint. The Amazon Bedrock API key (bearer token) goes in api_key; only the region/endpoint lives here.

class OpenAIBedrockKwargs(TypedDict, total=False):
    aws_region: str  # builds https://bedrock-runtime.<region>.amazonaws.com/openai/v1
    base_url: str    # or override the full endpoint URL (takes precedence)

AlibabaKwargs

class AlibabaKwargs(TypedDict, total=False):
    base_url: str        # override regional endpoint
    enable_search: bool  # enable Qwen's built-in web search

Regional endpoints:

• Singapore (default): https://dashscope-intl.aliyuncs.com/compatible-mode/v1
• Virginia (US): https://dashscope-us.aliyuncs.com/compatible-mode/v1
• Beijing (CN): https://dashscope.aliyuncs.com/compatible-mode/v1

See the DashScope model reference.

DeepSeekKwargs

class DeepSeekKwargs(TypedDict, total=False):
    base_url: str  # override the default DeepSeek API endpoint

GroqKwargs

class GroqKwargs(TypedDict, total=False):
    base_url: str  # must be the official Groq endpoint (https://api.groq.com/openai/v1)

Groq is bring-your-own-key only — register it with configure_provider("groq", api_key="gsk_..."); there is no VoiceRun-managed Groq key. See the Groq API reference.

voicerun

There is no VoicerunKwargs type and no voicerun key on the ProviderKwargs TypedDict, but provider_kwargs["voicerun"] is still read at runtime: any key you put there is forwarded verbatim to the upstream vLLM call (e.g. logprobs), with one exception — base_url is ignored. The serving endpoint is VoiceRun-owned server config, resolved per model from the session's provisioned model map, so a caller-supplied base_url never reaches the wire (on the completion path or the warm path).

This verbatim-passthrough of unrecognized keys is not unique to voicerun — it is inherited from the shared OpenAI-compatible base, so openai, alibaba, deepseek, and groq forward unknown provider_kwargs entries the same way. What is voicerun-specific is the suppressed base_url.

RetryConfiguration#

@dataclass
class RetryConfiguration:
    enabled: bool = True
    max_retries: int = 3
    retry_delay: float = 1.0          # initial delay in seconds
    backoff_multiplier: float = 2.0   # exponential backoff factor

StreamOptions#

@dataclass
class StreamOptions:
    chunk_by_sentence: bool = False
    clean_sentences: bool = True
    min_sentence_length: int = 6
    punctuation_marks: Optional[list[str]] = None
    punctuation_language: Optional[str] = None   # en | zh | ko | ja | es | fr | it | de

Deprecated: stream_sentences is the previous name for chunk_by_sentence. It still works for backward compatibility but emits a deprecation warning via primfunctions.logger.

Response types#

ChatCompletionResponse#

@dataclass
class ChatCompletionResponse:
    message: AssistantMessage
    finish_reason: str                           # "stop" | "length" | "tool_calls" | ...
    usage: Optional[dict[str, Any]] = None       # provider-native token counts
    provider: Optional[str] = None               # the provider that actually produced this response
    model: Optional[str] = None
    request_id: Optional[str] = None             # proxy-generated request id

provider and model reflect the actual executor — useful when fallbacks fire and you want to log which provider served the turn.

Message types#

Every message carries a vr_id field — an 8-character id auto-assigned on construction. It's preserved through serialize_conversation / deserialize_conversation so you can track individual messages across storage round-trips. You generally don't need to set it yourself.

UserMessage#

@dataclass
class UserMessage:
    content: str
    vr_id: str                                    # auto-generated
    name: Optional[str] = None
    cache_breakpoint: Optional[CacheBreakpoint] = None

AssistantMessage#

@dataclass
class AssistantMessage:
    content: Optional[str] = None
    vr_id: str                                    # auto-generated
    tool_calls: Optional[list[ToolCall]] = None
    cache_breakpoint: Optional[CacheBreakpoint] = None
    thought_signature: Optional[bytes] = None     # Google only; preserved automatically

SystemMessage#

Multiple SystemMessages are collapsed into one system block for Anthropic and Google.

@dataclass
class SystemMessage:
    content: str
    vr_id: str                                    # auto-generated
    cache_breakpoint: Optional[CacheBreakpoint] = None

ToolResultMessage#

The result of a tool call, fed back to the model on the next turn.

@dataclass
class ToolResultMessage:
    tool_call_id: str                             # matches ToolCall.id
    content: dict[str, Any]                       # JSON-serializable result
    vr_id: str                                    # auto-generated
    name: Optional[str] = None                    # function name
    cache_breakpoint: Optional[CacheBreakpoint] = None

ConversationHistory#

ConversationHistory = list[ConversationHistoryMessage]

A ConversationHistoryMessage is the union UserMessage | AssistantMessage | SystemMessage | ToolResultMessage.

Tool types#

ToolDefinition#

@dataclass
class ToolDefinition:
    type: Literal["function"]
    function: FunctionDefinition
    cache_breakpoint: Optional[CacheBreakpoint] = None

FunctionDefinition#

@dataclass
class FunctionDefinition:
    name: str
    description: str
    parameters: dict[str, Any]                    # JSON Schema
    strict: Optional[bool] = None                 # OpenAI strict-mode toggle

ToolCall#

Emitted by the model. Use .id when building the corresponding ToolResultMessage on the next turn.

@dataclass
class ToolCall:
    id: str
    type: Literal["function"]
    function: FunctionCall
    index: Optional[int] = None
    thought_signature: Optional[bytes] = None     # Google only

FunctionCall#

@dataclass
class FunctionCall:
    name: str
    arguments: dict[str, Any]                     # already JSON-parsed

ToolChoice#

ToolChoice = Union[Literal["none", "auto", "required"], str]

• "auto" — model decides (default)
• "none" — no tool calls
• "required" — must call at least one tool
• "<function_name>" — must call that specific function

Streaming types#

Every chunk has a .type: str property that lets you match on shape without importing every class:

async for chunk in stream:
    match chunk.type:
        case "content_delta":    ...   # ContentDeltaChunk
        case "content_sentence": ...   # ContentSentenceChunk
        case "tool_call":        ...   # ToolCallChunk
        case "finish_reason":    ...   # FinishReasonChunk
        case "usage":            ...   # UsageChunk
        case "response":         ...   # FinalResponseChunk
        case "error":            ...   # ErrorChunk (surfaces as CompletionsProxyError)

ChatCompletionChunk#

ChatCompletionChunk = Union[
    ContentDeltaChunk,
    ContentSentenceChunk,
    ToolCallChunk,
    FinishReasonChunk,
    UsageChunk,
    FinalResponseChunk,
    ErrorChunk,
]

ContentDeltaChunk#

Emitted when chunk_by_sentence=False. One per incremental token.

@dataclass
class ContentDeltaChunk:
    delta: str
    # type == "content_delta"

ContentSentenceChunk#

Emitted when chunk_by_sentence=True. One per complete sentence.

@dataclass
class ContentSentenceChunk:
    sentence: str
    # type == "content_sentence"

ToolCallChunk#

A fully-reassembled tool call. The proxy stitches together the streamed function-name + argument deltas before yielding.

@dataclass
class ToolCallChunk:
    tool_call: ToolCall
    # type == "tool_call"

FinishReasonChunk#

@dataclass
class FinishReasonChunk:
    finish_reason: str    # "stop" | "length" | "tool_calls" | ...
    # type == "finish_reason"

UsageChunk#

@dataclass
class UsageChunk:
    usage: dict[str, Any]
    # type == "usage"

FinalResponseChunk#

The last chunk of a successful stream. Carries the fully-assembled ChatCompletionResponse.

@dataclass
class FinalResponseChunk:
    response: ChatCompletionResponse
    # type == "response"

ErrorChunk#

Emitted by the proxy on a mid-stream failure. The library re-raises it as CompletionsProxyError inside async for, so you should not need to match it explicitly.

@dataclass
class ErrorChunk:
    error: str
    error_type: str
    # type == "error"

Enums#

CompletionsProvider#

class CompletionsProvider(StrEnum):
    OPENAI            = "openai"
    OPENAI_RESPONSES  = "openai_responses"
    ANTHROPIC         = "anthropic"
    GOOGLE            = "google"
    GOOGLE_VERTEX     = "google_vertex"
    ANTHROPIC_VERTEX  = "anthropic_vertex"
    ANTHROPIC_BEDROCK = "anthropic_bedrock"
    OPENAI_BEDROCK    = "openai_bedrock"
    ALIBABA           = "alibaba"
    DEEPSEEK          = "deepseek"
    GROQ              = "groq"
    VOICERUN          = "voicerun"

Since it's a StrEnum, CompletionsProvider.GOOGLE == "google". Pass either form wherever a provider is expected. These twelve values are the complete set — the proxy rejects any other provider string, which reaches your handler as CompletionsProxyError with error_type == "InvalidProviderError".

CompletionsProvider.OPENAI_RESPONSES routes to OpenAI's Responses API (/v1/responses) instead of chat completions — see OpenAI Responses API.

CompletionsProvider.VOICERUN targets VoiceRun's self-hosted, OpenAI-compatible vLLM models — for example qwen3-0.6b-slm, the current client-facing name for that deployment and the id to use. (qwen3-0.6b-inline-slot-v5 is its pre-rename name, kept only so traffic still in flight at the rename cutover keeps billing correctly; it is expected to be retired once nothing references it.) The proxy supplies the endpoint and key when this provider is configured for the session, so there is no bring-your-own-key form: register it with configure_provider("voicerun", voicerun_managed=True). The serving endpoint is resolved server-side per model from the session's provisioned model map, falling back to the cluster default. If neither the model map nor the cluster default resolves an endpoint, the request raises a configuration error rather than falling through to a public endpoint.

Cache types#

CacheBreakpoint#

@dataclass
class CacheBreakpoint:
    ttl: Literal["5m", "1h"] = "5m"

See Advanced features → Anthropic cache breakpoints for usage rules.

Utility functions#

serialize_conversation#

def serialize_conversation(conversation: ConversationHistory) -> list[dict[str, Any]]

deserialize_conversation#

def deserialize_conversation(data: list[dict[str, Any]]) -> ConversationHistory

Use these when round-tripping conversation history through storage that wants plain dicts (e.g. context.set_completion_messages / context.get_completion_messages).