docs.VoiceRun

Basic Usage

Simple chat completion#

Register the provider once in StartEvent, then call generate_chat_completion on each turn.

from primfunctions.events import Event, StartEvent, TextEvent, TextToSpeechEvent
from primfunctions.context import Context
from primfunctions.completions import (
    configure_provider,
    generate_chat_completion,
)


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

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

        response = await generate_chat_completion({
            "provider": "anthropic",
            "model": "claude-haiku-4-5",
            "messages": [{"role": "user", "content": user_message}],
        })

        if response.message.content:
            yield TextToSpeechEvent(
                text=response.message.content,
                voice="kore",
            )

Note the shape:

  • api_key is not in the request body. It was registered via configure_provider and lives in the session's provider map.
  • Messages can be dicts (shown above) or typed dataclasses (shown below). Both are accepted.

Typed request object#

For better type safety, build the request with the dataclass forms:

from primfunctions.completions import (
    ChatCompletionRequest,
    CompletionsProvider,
    UserMessage,
    configure_provider,
    generate_chat_completion,
)

async def handler(event, context):
    if isinstance(event, StartEvent):
        configure_provider("anthropic", voicerun_managed=True)
        yield TextToSpeechEvent(text="Hello!", voice="kore")

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

        request = ChatCompletionRequest(
            provider=CompletionsProvider.ANTHROPIC,
            model="claude-haiku-4-5",
            messages=[UserMessage(content=user_message)],
        )

        response = await generate_chat_completion(request)
        if response.message.content:
            yield TextToSpeechEvent(
                text=response.message.content,
                voice="kore",
            )

CompletionsProvider is a StrEnum — CompletionsProvider.ANTHROPIC == "anthropic" — so you can pass either form wherever provider is expected.

Multi-turn conversations#

primfunctions.context.Context has three helpers for persisting conversation state across turns. They accept either dicts or typed message dataclasses.

from primfunctions.completions import (
    ConversationHistory,
    UserMessage,
    configure_provider,
    deserialize_conversation,
    generate_chat_completion,
)


async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        configure_provider("anthropic", voicerun_managed=True)
        yield TextToSpeechEvent(
            text="Hello! I can remember our conversation.",
            voice="kore",
        )

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

        # Pull stored history back into typed objects
        messages: ConversationHistory = deserialize_conversation(
            context.get_completion_messages()
        )
        messages.append(UserMessage(content=user_message))

        response = await generate_chat_completion({
            "provider": "anthropic",
            "model": "claude-haiku-4-5",
            "messages": messages,
        })

        messages.append(response.message)
        context.set_completion_messages(messages)

        if response.message.content:
            yield TextToSpeechEvent(
                text=response.message.content,
                voice="kore",
            )

Context completion-message helpers#

The Context object exposes three helpers for managing the conversation history that primfunctions.completions consumes:

context.get_completion_messages()#

Returns the conversation history as a list of dicts. Use deserialize_conversation() to convert to typed message objects.

from primfunctions.completions import deserialize_conversation

raw = context.get_completion_messages()  # list[dict]
messages = deserialize_conversation(raw)  # list[ConversationHistoryMessage]

context.add_completion_message(message)#

Appends a single message. Accepts a typed dataclass (UserMessage, AssistantMessage, SystemMessage, ToolResultMessage) or a raw dict.

from primfunctions.completions import UserMessage

context.add_completion_message(UserMessage(content="Hello"))
context.add_completion_message(response.message)  # AssistantMessage
context.add_completion_message({"role": "user", "content": "Hello"})  # dict

context.set_completion_messages(messages)#

Replaces the entire stored history. Accepts a list that can mix typed messages and dicts.

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

Recommended pattern: write once, at end of turn#

If you call add_completion_message incrementally, an interrupted turn can leave the stored history half-written (user message present, assistant response missing, tool result dangling). Preferring one set_completion_messages at the end of the turn keeps the history atomic:

# 1. Load existing history
messages = deserialize_conversation(context.get_completion_messages())

# 2. Build the turn's worth of new messages locally
messages.append(UserMessage(content=user_message))
response = await generate_chat_completion({...})
messages.append(response.message)

# 3. Commit once
context.set_completion_messages(messages)

System messages#

from primfunctions.completions import (
    SystemMessage,
    UserMessage,
    configure_provider,
    generate_chat_completion,
)


async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        configure_provider("anthropic", voicerun_managed=True)
        yield TextToSpeechEvent(
            text="Ahoy! I be a helpful assistant that speaks like a pirate.",
            voice="kore",
        )

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

        response = await generate_chat_completion({
            "provider": "anthropic",
            "model": "claude-haiku-4-5",
            "messages": [
                SystemMessage(content="You are a helpful assistant that speaks like a pirate."),
                UserMessage(content=user_message),
            ],
        })

        if response.message.content:
            yield TextToSpeechEvent(
                text=response.message.content,
                voice="kore",
            )

Request parameters#

Common scalars on ChatCompletionRequest:

response = await generate_chat_completion({
    "provider": "anthropic",
    "model": "claude-haiku-4-5",
    "messages": [{"role": "user", "content": user_message}],
    "temperature": 0.7,   # 0.0–1.0, controls randomness
    "max_tokens": 500,    # upper bound on output length
    "timeout": 30.0,      # per-attempt timeout in seconds
})

See the API Reference for the full field list.

Next steps#

chatcompletiongetting-started