voicerun_completions

A comprehensive, unified SDK for interacting with multiple LLM providers (OpenAI, Anthropic, Google, and Vertex Anthropic) through a single, consistent API. This library simplifies working with different providers by normalizing their APIs and providing powerful features like automatic fallbacks, retry logic, and advanced streaming options.

Quick Start

from primfunctions.events import Event, StartEvent, TextEvent, TextToSpeechEvent
from primfunctions.context import Context
from voicerun_completions import generate_chat_completion

async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        yield TextToSpeechEvent(
            text="Hello! Ask me anything and I'll answer your question.",
            voice="nova"
        )

    if isinstance(event, TextEvent):
        user_message = event.data.get("text", "")
        
        response = await generate_chat_completion({
            "provider": "openai",
            "api_key": context.variables.get("OPENAI_API_KEY"),
            "model": "gpt-5-mini",
            "messages": [
                {"role": "user", "content": user_message}
            ]
        })
        
        if response.message.content:
            yield TextToSpeechEvent(
                text=response.message.content,
                voice="nova"
            )

Connection Reuse with CompletionsClient

Use CompletionsClient to reuse HTTP connections across requests. This avoids the TLS handshake overhead of creating a new SDK client on every call — important for voice agents where latency matters.

In a VoiceRun Agent

Create the client at module level so it persists across handler invocations. The agent handler is called per-event, but the module-level client lives for the lifetime of the process, keeping connections warm.

from primfunctions.events import Event, StartEvent, TextEvent, TextToSpeechEvent
from primfunctions.context import Context
from voicerun_completions import CompletionsClient, deserialize_conversation, UserMessage

# Module-level client — connections are reused across all handler calls
completions = CompletionsClient()

async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        yield TextToSpeechEvent(text="Hello! Ask me anything.", voice="nova")

    if isinstance(event, TextEvent):
        user_message = event.data.get("text", "N/A")

        messages = deserialize_conversation(context.get_completion_messages())
        messages.append(UserMessage(content=user_message))

        stream = await completions.generate_chat_completion_stream(
            request={
                "provider": "anthropic",
                "api_key": context.variables.get("ANTHROPIC_API_KEY"),
                "model": "claude-haiku-4-5",
                "messages": messages,
            },
            stream_options={"stream_sentences": True, "clean_sentences": True},
        )

        async for chunk in stream:
            if chunk.type == "content_sentence":
                yield TextToSpeechEvent(text=chunk.sentence, voice="nova")
            elif chunk.type == "response":
                messages.append(chunk.response.message)
                context.set_completion_messages(messages)

In Other Applications

For servers or scripts where you control the lifecycle, you can also use the context manager:

from voicerun_completions import CompletionsClient

async with CompletionsClient() as client:
    response = await client.generate_chat_completion({...})
    stream = await client.generate_chat_completion_stream({...})
# All SDK connections closed automatically

The module-level generate_chat_completion() and generate_chat_completion_stream() functions still work as before without connection reuse.

Documentation

• Installation - Installation instructions and prerequisites
• Core Concepts - Providers, message types, and request formats
• Basic Usage - Getting started with chat completions
• Streaming - Token and sentence-based streaming
• Tool/Function Calling - Using tools and function calling
• Reliability: Retries & Fallbacks - Retry logic and automatic fallbacks
• Advanced Features - Cache breakpoints, vendor-specific options, and more
• API Reference - Complete API documentation
• Examples - Complete working examples
• Best Practices - Best practices and error handling

Testing

Install dev dependencies:

uv pip install -e ".[dev]"

Run unit tests (no API keys required):

uv run pytest -m "not integration" -v

Run all tests including integration (requires API keys in environment):

uv run pytest -v

Run a specific test file:

uv run pytest tests/test_google_schema_sanitization.py -v

Integration tests require one or more of these environment variables:

• OPENAI_API_KEY
• ANTHROPIC_API_KEY
• GEMINI_API_KEY
• GCP_SERVICE_ACCOUNT_JSON / GCP_PROJECT_ID (for Vertex tests)

Tests missing the required key will be skipped automatically.

Features

• Unified API - Single interface for OpenAI, Anthropic, Google, and Vertex Anthropic
• Automatic Fallbacks - Seamlessly fallback to alternative providers on failure
• Retry Logic - Built-in exponential backoff retry mechanism
• Advanced Streaming - Token-based and sentence-based streaming for real-time applications
• Tool Calling - Unified tool/function calling across all providers
• Type Safety - Full type hints and support for both dict and object formats

Git Worktrees (Parallel Agent Development)

This repo supports running multiple coding agents in parallel using git worktrees. Each agent gets an isolated working directory with its own branch while sharing the same git object store.

Worktrees are stored in .worktrees/ (gitignored).

Create a worktree

git worktree add .worktrees/feature-xyz -b feature-xyz

List active worktrees

git worktree list

Remove a worktree

git worktree remove .worktrees/feature-xyz

Clean up stale worktrees

git worktree prune