Tool / Function Calling

primfunctions.completions presents one tool-calling interface that works across every provider. You describe tools as JSON Schema, the model emits tool calls, you execute them and feed the results back on the next turn.

Basic tool call#

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="I can check the weather. Ask me about any location.",
            voice="kore",
        )

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

        tools = [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "Get the current weather for a location",
                    "parameters": {
                        "type": "object",
                        "properties": {
                            "location": {
                                "type": "string",
                                "description": "City and state, e.g. San Francisco, CA",
                            },
                            "unit": {
                                "type": "string",
                                "enum": ["celsius", "fahrenheit"],
                            },
                        },
                        "required": ["location"],
                    },
                },
            }
        ]

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

        if response.message.tool_calls:
            for tool_call in response.message.tool_calls:
                yield TextToSpeechEvent(
                    text=f"Looking up {tool_call.function.arguments}",
                    voice="kore",
                )
        elif response.message.content:
            yield TextToSpeechEvent(text=response.message.content, voice="kore")

Typed tool definitions#

ToolDefinition and FunctionDefinition give you type checking and enum autocomplete for fields like strict:

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


async def handler(event, context):
    if isinstance(event, StartEvent):
        configure_provider("anthropic", voicerun_managed=True)

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

        tools = [
            ToolDefinition(
                type="function",
                function=FunctionDefinition(
                    name="get_weather",
                    description="Get the current weather for a location",
                    parameters={
                        "type": "object",
                        "properties": {
                            "location": {"type": "string"},
                            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
                        },
                        "required": ["location"],
                    },
                ),
            ),
        ]

        request = ChatCompletionRequest(
            provider=CompletionsProvider.ANTHROPIC,
            model="claude-haiku-4-5",
            messages=[UserMessage(content=user_message)],
            tools=tools,
            tool_choice="auto",
        )
        response = await generate_chat_completion(request)
        ...

Dicts and typed ToolDefinition objects are interchangeable on tools and on tool_choice — use whichever reads better in your codebase.

Full tool-execution loop#

A realistic turn:

1. Call the model with tools
2. Execute each response.message.tool_calls[*] locally
3. Append the assistant's tool-call message and a ToolResultMessage per tool call to the history
4. Call the model again — it sees the tool results and produces the final answer

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


async def get_weather(location: str) -> dict:
    # Your actual backend call
    return {"temperature": 72, "condition": "sunny"}


async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        configure_provider("anthropic", voicerun_managed=True)
        yield TextToSpeechEvent(
            text="I can check the weather for multiple locations.",
            voice="kore",
        )

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

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

        tools = [
            {
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "description": "Get weather for a location",
                    "parameters": {
                        "type": "object",
                        "properties": {"location": {"type": "string"}},
                        "required": ["location"],
                    },
                },
            },
        ]

        # First call — model may ask for a tool
        response = await generate_chat_completion({
            "provider": "anthropic",
            "model": "claude-haiku-4-5",
            "messages": messages,
            "tools": tools,
            "tool_choice": "auto",
        })
        messages.append(response.message)

        # Execute any tool calls and feed results back
        if response.message.tool_calls:
            for tool_call in response.message.tool_calls:
                result = await get_weather(tool_call.function.arguments["location"])
                messages.append(ToolResultMessage(
                    tool_call_id=tool_call.id,
                    name=tool_call.function.name,
                    content=result,
                ))

            # Second call — model uses the tool results
            response = await generate_chat_completion({
                "provider": "anthropic",
                "model": "claude-haiku-4-5",
                "messages": messages,
                "tools": tools,
                "tool_choice": "auto",
            })
            messages.append(response.message)

        context.set_completion_messages(messages)

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

For more than one tool, wrap the bodies of get_weather (and friends) in your own dispatch function and have it return a ToolResultMessage.

Streaming with tool calls#

Tool calls arrive as a single ToolCallChunk — the proxy reassembles the streamed function-name + argument deltas for you before yielding.

from primfunctions.completions import (
    configure_provider,
    generate_chat_completion_stream,
)


async def handler(event, context):
    if isinstance(event, StartEvent):
        configure_provider("anthropic", voicerun_managed=True)

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

        stream = await generate_chat_completion_stream(
            request={
                "provider": "anthropic",
                "model": "claude-haiku-4-5",
                "messages": [{"role": "user", "content": user_message}],
                "tools": [
                    {
                        "type": "function",
                        "function": {
                            "name": "get_weather",
                            "description": "Get weather for a location",
                            "parameters": {
                                "type": "object",
                                "properties": {"location": {"type": "string"}},
                                "required": ["location"],
                            },
                        },
                    },
                ],
                "tool_choice": "auto",
            },
            stream_options={"chunk_by_sentence": True, "clean_sentences": True},
        )

        async for chunk in stream:
            if chunk.type == "content_sentence":
                yield TextToSpeechEvent(text=chunk.sentence, voice="kore")
            elif chunk.type == "tool_call":
                # chunk.tool_call.id / .function.name / .function.arguments
                pass
            elif chunk.type == "response":
                final = chunk.response

tool_choice options#

OpenAI strict mode#

OpenAI supports a strict: true flag on a function definition that forces argument validation against the schema. Set it on the nested FunctionDefinition (or the function dict):

tools = [
    ToolDefinition(
        type="function",
        function=FunctionDefinition(
            name="get_weather",
            description="Get weather for a location",
            parameters={...},
            strict=True,  # OpenAI only
        ),
    ),
]

Schema portability#

Tool parameters accepts any valid JSON Schema. Not every provider supports every JSON Schema feature — Google in particular has a narrower surface.

See JSON Schema support for the cross-provider compatibility matrix, Google's automatic sanitization behavior, and recommendations for maximum portability.

Next steps#

• Streaming — more on streaming chunk types including ToolCallChunk
• Advanced features — cache breakpoints on tools
• Examples — a full multi-turn tool-calling handler