Context Reference

The Context object manages session state, configuration, and provides utilities for data management, background tasks, A/B testing, and outcome tracking.

Properties#

class Context:
    agent_id: str           # Unique identifier for the agent
    environment: str        # Environment name and ID (e.g., "production|a1b2c3d4-...")
    session_id: str         # Unique session identifier
    function_id: str        # Current function identifier
    variables: dict         # Environment and organization variables
    data: dict              # Custom session data store
    testing: Testing        # Testing and outcomes manager

Usage:

from primfunctions.logger import logger

async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        # Access session info
        logger.info(f"Agent: {context.agent_id}")
        logger.info(f"Environment: {context.environment}")
        logger.info(f"Session: {context.session_id}")

        # Access environment variables (API keys, config)
        api_key = context.variables.get("OPENAI_API_KEY")

Variables#

context.variables is a plain dict[str, str] populated when the agent session starts. It holds configuration values — API keys, endpoints, feature flags — that you can read from anywhere inside your handler with standard dict access.

api_key = context.variables.get("ANTHROPIC_API_KEY")
region = context.variables.get("AWS_REGION", "us-east-1")  # with default

Where variables come from#

Variables are merged from two scopes when the session starts:

If a variable with the same name exists at both scopes, the environment-scoped value wins.

Keys are case-sensitive. context.variables.get("API_KEY") will not return a variable created as api_key.

Creating variables#

Use the VoiceRun CLI to create variables:

# Organization-level (every agent can read it)
vr create variable SUPPORT_EMAIL help@acme.com --org

# Environment-scoped, visible only to sessions in the "production" environment
vr create variable GREETING_VOICE nova \
    --agent my-agent --environment production

Inside a voicerun project, the --agent flag defaults to the agent in .voicerun/agent.lock, so you can usually omit it:

vr create variable GREETING_VOICE nova --environment production

Variables are for non-sensitive configuration. For API keys, tokens, and other credentials, use secrets instead — see Variables vs. secrets below.

Variables vs. secrets#

VoiceRun also exposes organization-scoped secrets (vr create secret, OrganizationSecret), and they are runtime-injectable into context.variables — reference them from your agent's configuration with a placeholder:

# .voicerun/values.yaml (or directly in a template)
variables:
  ANTHROPIC_API_KEY: "{{ Secrets.organization.ANTHROPIC_API_KEY }}"

The placeholder is plain data to Helm, so it survives rendering verbatim into the release manifest. At session start the platform scans the manifest for {{ Secrets.organization.<NAME> }} references, resolves only the referenced names in-process, and merges the values into context.variables — so your handler reads them exactly like any variable:

api_key = context.variables.get("ANTHROPIC_API_KEY")

Store sensitive values — API keys, tokens, credentials — as secrets. Values never appear in the stored manifest, the wire parameters, or logs, and are resolved in-process only at session start. A missing secret degrades gracefully: context.variables.get(NAME) returns None. Use plain variables for non-sensitive configuration.

Secrets are also available to evaluators during test and evaluation workflows.

Data Management#

Store and retrieve session-specific data that persists throughout the conversation.

set_data#

def set_data(self, key: str, value: Any)

Set the value of a data key.

get_data#

def get_data(self, key: str, default: Any = None) -> Any

Get the value of a data key, with optional default.

Usage:

# Store session data
context.set_data("user_name", "John")
context.set_data("order_items", ["coffee", "sandwich"])
context.set_data("order_total", 15.99)

# Retrieve session data
user_name = context.get_data("user_name", "Guest")
items = context.get_data("order_items", [])

Subagent Input & Output#

Agents can be invoked as subagents: the caller hands the platform a task spec when the session is created, the handler runs autonomously, and a structured result is persisted back onto the session at session end. Use this pattern when an outbound call (or any pre-created session) needs to return a structured outcome to whatever kicked it off — for example, "did the booking get confirmed, and what's the confirmation number?"

The two surfaces involved:

Both fields are JSON objects capped at 64KB. Top-level value must be a dict.

input_data#

context.input_data: dict

Read-only snapshot of the JSON the caller passed in when the session was created. Always a dict — empty when no payload was provided.

set_output#

def set_output(self, payload: dict) -> None

Replace the output payload. Raises TypeError if payload is not a dict and ValueError if it's not JSON-serializable or exceeds the 64KB cap.

To build the output incrementally across turns, spread the current value into the new payload:

context.set_output({**context.output_data, "field": value})

Usage:

from primfunctions.events import StartEvent, StopEvent, TextEvent, TextToSpeechEvent
from primfunctions.logger import logger


async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        # Read the task spec the caller provided.
        task = context.input_data
        logger.info(f"Running outbound task: {task.get('objective')}")
        yield TextToSpeechEvent(text="Hi, I'm calling about your booking.", voice="nova")

    if isinstance(event, TextEvent):
        # ... drive the conversation, extracting fields as you go ...
        if confirmed_a_thing:
            context.set_output({
                **context.output_data,
                "status": "confirmed",
                "confirmation_number": "ABC123",
            })

    if isinstance(event, StopEvent):
        # output_data has already been built up; the platform persists it
        # onto the session at session end. No explicit "send" is needed.
        logger.info(f"Final output: {context.output_data}")

The caller picks up the result either by polling vr session info for the outputData field, by passing --wait to vr outbound call, or — once the session-end webhook is extended — by receiving it as part of the webhook payload.

Persistence semantics: the platform reads context.output_data once, at session end, and writes it onto the session record. Multiple set_output calls during the call are fine; only the final value is persisted.

Size cap: 64KB enforced eagerly at the SDK and again at the API. Use it for structured results, not transcripts (those have their own field).

Shared Data#

context.shared_data reads JSON payloads that an external system stored for your organization under a key you choose.

Use it when data arrives beside a call rather than on it — an upstream system delivers the call carrying a short correlation id, and sends the payload that belongs to it separately. input_data above cannot help there: it is fixed at session creation, and the payload may not have arrived yet.

async def get(key: str, default: Any = None) -> Any

correlation_id = context.get_data("transaction_id")
payload = await context.shared_data.get(f"lead:{correlation_id}")
if payload:
    for field, value in payload.items():
        context.set_data(field, value)

Every kind of miss returns default — never written, expired, pinned to a different agent, or a transport failure. Handlers must be able to carry on without the payload; delivery from an external system is usually at-most-once.

Entries are organization-scoped unless the writer pinned them to one agent, so any session in your organization that knows the key can read an unpinned entry.

See Shared Data Store for the write endpoint, key and TTL rules, agent pinning, and a handler pattern that degrades correctly when the payload is late or absent.

Completion Messages#

Manage conversation history for use with LLM completions via VoiceRun Completions (primfunctions.completions).

get_completion_messages#

def get_completion_messages(self) -> list[dict[str, Any]]

Return the stored conversation history as a list of dicts. Use primfunctions.completions.deserialize_conversation to convert to typed message objects.

add_completion_message#

def add_completion_message(self, message: Union[SerializableMessage, dict[str, Any]])

Append a message to the history. Accepts any primfunctions.completions message dataclass (UserMessage, AssistantMessage, SystemMessage, ToolResultMessage) or a plain dict.

set_completion_messages#

def set_completion_messages(self, messages: list[Union[SerializableMessage, dict[str, Any]]])

Replace the entire history. The list may mix typed messages and dicts.

Usage:

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


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

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

        # Get existing messages and add new user message
        messages = deserialize_conversation(context.get_completion_messages())
        messages.append(UserMessage(content=user_message))

        # Generate response
        response = await generate_chat_completion({
            "provider": "google",
            "model": "gemini-3.5-flash",
            "messages": messages,
        })

        # Store updated conversation
        messages.append(response.message)
        context.set_completion_messages(messages)

        yield TextToSpeechEvent(text=response.message.content, voice="nova")

Conversation Transcript#

context.get_transcript() returns a running, chronological record of what was actually said and heard during the call. The platform builds it for you turn by turn — you don't append to it. Each entry is a dict:

get_transcript#

def get_transcript(self) -> list[dict[str, Any]]

Return the running transcript as a list of {"role", "content", "interrupted"} dicts. The returned list is a defensive copy — mutating it does not change the stored transcript. The read-only context.transcript property returns the same thing.

Heard-accurate assistant turns#

The transcript's assistant entries reflect what the caller actually heard, not just what the agent intended to say. When a caller barges in (interrupts the agent mid-sentence), the assistant entry is truncated to the portion that played before the interruption and flagged interrupted: True.

That "heard prefix" comes from the platform's interruption transcription.

A turn the agent finished without interruption keeps its full text and interrupted: False. A turn the caller cut off so early that nothing meaningful was heard is dropped from the transcript entirely.

Usage:

from primfunctions.logger import logger


async def handler(event: Event, context: Context):
    if isinstance(event, TextEvent):
        # Read the heard-accurate record of the call so far.
        for turn in context.get_transcript():
            suffix = " (interrupted)" if turn["interrupted"] else ""
            logger.info(f"[{turn['role']}] {turn['content']}{suffix}")

After a caller interrupts the greeting, context.get_transcript() looks like:

[
    {
        "role": "assistant",
        "content": "Hello and thank you for calling Riverside Bistro, my name is",
        "interrupted": True,
    },
    {
        "role": "user",
        "content": "I need a table for four tonight at seven",
        "interrupted": False,
    },
]

Notice the assistant's greeting is truncated to the words the caller heard before cutting in — the rest of the sentence the agent never got to say is not in the record.

Using the transcript with completions#

The transcript entries are already valid completion messages, so you can pass get_transcript() straight into generate_chat_completion as the messages — typically with a system prompt prepended:

from primfunctions.completions import configure_provider, generate_chat_completion


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

    if isinstance(event, TextEvent):
        # The heard transcript IS the message history — no conversion needed.
        messages = [
            {"role": "system", "content": "You are a friendly restaurant host."},
            *context.get_transcript(),
        ]

        response = await generate_chat_completion({
            "provider": "openai",
            "model": "gpt-4o-mini",
            "messages": messages,
        })

        yield TextToSpeechEvent(text=response.message.content, voice="nova")

Each {"role", "content"} entry maps to a user or assistant message, the extra interrupted field is ignored, and a vr_id is filled in automatically.

Preferred for simple agents: feeding get_transcript() directly to generate_chat_completion is the recommended pattern for simple agents that don't use tool calling and don't emit intermediary assistant responses (assistant turns that were never spoken to the caller, e.g. silent tool-result steps). The transcript already holds the heard-accurate user/assistant turns, so there's nothing to assemble or prune. If your agent calls tools, injects system-role steering mid-conversation, or produces assistant messages that weren't spoken aloud, use get_completion_messages() instead — it preserves tool calls, tool results, and every message the model was given.

Reading it elsewhere: get_transcript() is available inside the handler during the live session. To retrieve a session's transcript after it ends, use the Transcripts API.

Trimming history to what the caller heard#

When you manage your own conversation history (for tool calling, system steering, or anything beyond the simple pass-through above), a barge-in leaves the model's record out of sync with reality: the trailing assistant message holds everything you sent to TTS, but the caller only heard part of it. Left unchanged, the model assumes it communicated things it never got to say.

truncate_to_heard_prefix(text, heard) trims a written assistant message to the spoken prefix the caller heard. It aligns the two on words, so number/currency normalization ("$8.99" vs "eight dollars and ninety-nine cents") doesn't desync the cut. Drive it off the transcript's last assistant entry — the heard-accurate record from above:

from primfunctions.completions import AssistantMessage, deserialize_conversation
from primfunctions.transcript import truncate_to_heard_prefix


async def handler(event: Event, context: Context):
    if isinstance(event, TextEvent):
        messages = deserialize_conversation(context.get_completion_messages())

        # If the caller barged in over the previous turn, trim the trailing
        # assistant message to what they actually heard.
        heard = next((t for t in reversed(context.get_transcript()) if t["role"] == "assistant"), None)
        if heard and heard["interrupted"] and messages and isinstance(messages[-1], AssistantMessage):
            cut = truncate_to_heard_prefix(messages[-1].content or "", heard["content"])
            if cut is not None:
                messages[-1].content = cut

        # ... append the new user message, call the model, persist messages ...

truncate_to_heard_prefix returns None when the heard text is too dissimilar to align confidently — leave the message unchanged (or apply your own conservative fallback) in that case. It only touches the single most recent assistant message; earlier turns are settled and aren't reconciled.

Cache#

Store and retrieve cached data that persists across the session.

cache_set#

def cache_set(self, key: str, value: Any)

Add a value to the cache. Supports CachableEntity objects, dicts, and primitives (str, int, float, bool, None).

cache_get#

def cache_get(self, key: str, entity_type: Optional[Type] = None) -> Optional[Any]

Get a value from the cache. If entity_type is provided, deserializes to that type.

Usage:

# Cache primitive values
context.cache_set("last_query", "weather in NYC")
context.cache_set("query_count", 5)

# Retrieve cached values
last_query = context.cache_get("last_query")
count = context.cache_get("query_count")

Background Tasks#

Create and manage background tasks for long-running operations that shouldn't block the conversation.

create_task#

def create_task(
    self,
    handler: Coroutine,
    name: str = None,
    interruptible: bool = False,
    timeout: int = 30,
)

Create a background task.

cancel_task#

def cancel_task(self, name: str)

Cancel a specific task by name.

cancel_interruptible_tasks#

def cancel_interruptible_tasks(self)

Cancel all tasks marked as interruptible.

cancel_all_tasks#

def cancel_all_tasks(self)

Cancel all background tasks.

has_unfinished_tasks#

def has_unfinished_tasks(self) -> bool

Check if any tasks are still running.

wait_for_all_tasks#

async def wait_for_all_tasks(self)

Wait for all tasks to complete.

Usage:

from primfunctions.logger import logger

async def process_order(context: Context):
    """Background task that yields events."""
    logger.info("Starting order processing...")

    # Simulate processing
    await asyncio.sleep(5)

    context.set_data("order_status", "completed")
    logger.info("Order processing complete")

async def handler(event: Event, context: Context):
    if isinstance(event, TextEvent):
        user_message = event.data.get("text", "N/A").lower()

        if "place order" in user_message:
            # Start background task
            context.create_task(
                process_order(context),
                name="order_processing",
                timeout=60
            )

            yield TextToSpeechEvent(
                text="Processing your order in the background.",
                voice="nova"
            )

        elif "check status" in user_message:
            status = context.get_data("order_status", "pending")
            yield TextToSpeechEvent(
                text=f"Your order status is: {status}",
                voice="nova"
            )

Tests (A/B Testing)#

Create randomized tests for experimenting with different conversation approaches.

add_test#

def add_test(self, name: str, options: dict, description: str = "", stop: dict = {}) -> str

Add a test and return its selected value.

Stop conditions:

get_test#

def get_test(self, name: str) -> str

Get the selected value for a test. Returns the same value throughout the session.

add_tests#

def add_tests(self, tests: list[dict])

Add multiple tests at once.

Usage:

async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        # Simple A/B test
        greeting = context.add_test(
            name="greeting_style",
            options={
                "Hello! How can I help you today?": 0.5,
                "Hi there! What can I do for you?": 0.3,
                "Welcome! How may I assist you?": 0.2
            }
        )

        yield TextToSpeechEvent(text=greeting, voice="nova")

    if isinstance(event, TextEvent):
        # Get the same greeting value (consistent within session)
        greeting = context.get_test("greeting_style")

With stop conditions:

context.add_test(
    name="upsell_approach",
    options={"direct": 0.5, "subtle": 0.5},
    stop={
        "max_iterations": 1000,
        "max_confidence": 95,
        "target_outcome": "conversion_rate",
        "default": "direct",
        "stop_on": 2,
        "notify": ["analytics@company.com"]
    }
)

Outcomes#

Track metrics and conversion events for analytics. Outcomes are linked to tests for A/B analysis.

add_outcome#

def add_outcome(self, name: str, type: str, description: str = "")

Add an outcome to track.

get_outcome#

def get_outcome(self, name: str) -> Any

Get the current value of an outcome.

set_outcome#

def set_outcome(self, name: str, value: Any) -> Any

Set the value of an outcome. Returns the previous value.

trigger_outcome#

def trigger_outcome(self, name: str) -> Any

Trigger an outcome. For booleans, sets to True. For integers, increments by 1.

reset_outcome#

def reset_outcome(self, name: str) -> Any

Reset an outcome to its default value. Returns the previous value.

add_outcomes#

def add_outcomes(self, outcomes: list[dict])

Add multiple outcomes at once.

Usage:

async def handler(event: Event, context: Context):
    if isinstance(event, StartEvent):
        # Define outcomes
        context.add_outcome("message_count", "integer", "Total messages")
        context.add_outcome("user_satisfied", "boolean", "User expressed satisfaction")
        context.add_outcome("order_value", "float", "Total order value")

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

        # Increment counter
        context.trigger_outcome("message_count")

        # Set boolean
        if "thank you" in user_message or "great" in user_message:
            context.trigger_outcome("user_satisfied")

        # Set numeric value
        if "order" in user_message:
            context.set_outcome("order_value", 29.99)

        # Get current values
        count = context.get_outcome("message_count")
        yield TextToSpeechEvent(
            text=f"This is message number {count}.",
            voice="nova"
        )

Testing Metadata#

Store arbitrary metadata associated with the testing session.

set_testing_metadata#

def set_testing_metadata(self, key: str, value: Any)

Set a metadata value.

get_testing_metadata#

def get_testing_metadata(self, key: str, default: Any = None) -> Any

Get a metadata value.

Usage:

# Store metadata about the test session
context.set_testing_metadata("segment", "returning_customer")
context.set_testing_metadata("source", "phone")

# Retrieve metadata
segment = context.get_testing_metadata("segment")