hu-sdk 0.0.1

Python SDK for building voice agents on the Voice Gateway

Voice Agent SDK (Python)

Python SDK for building voice agents on the Voice Gateway.

Installation

pip install voice-agent

Quick Start

import asyncio
from voice_agent import VoiceAgent, VoiceAgentConfig, ConnectionMode

config = VoiceAgentConfig(
    api_key="sk-voice-xxx",
    gateway_url="wss://gateway.example.com",
    mode=ConnectionMode.WEBSOCKET,
)

agent = VoiceAgent(config)

@agent.on_utterance
async def handle_utterance(ctx):
    print(f"User said: {ctx.text}")

    # Stream response
    ctx.send_delta("Hello ")
    ctx.send_delta("World!")
    ctx.done()

@agent.on_interrupt
def handle_interrupt(session_id: str, reason: str):
    print(f"Interrupted: {reason}")

@agent.on_error
def handle_error(error: Exception):
    print(f"Error: {error}")

async def main():
    await agent.connect()
    # Keep running
    while agent.is_connected():
        await asyncio.sleep(1)

asyncio.run(main())

Streaming with LLM

import asyncio
from openai import AsyncOpenAI
from voice_agent import VoiceAgent, VoiceAgentConfig

openai = AsyncOpenAI()

config = VoiceAgentConfig(
    api_key=os.environ["VOICE_API_KEY"],
    gateway_url=os.environ["GATEWAY_URL"],
)

agent = VoiceAgent(config)

@agent.on_utterance
async def handle(ctx):
    stream = await openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": ctx.text}],
        stream=True,
    )

    async for chunk in stream:
        if ctx.is_aborted:
            break

        delta = chunk.choices[0].delta.content
        if delta:
            ctx.send_delta(delta)

    ctx.done()

asyncio.run(agent.connect())

Using Vision (Video Frames)

Agents with vision scope can request video frames:

from voice_agent import FrameRequestOptions

@agent.on_utterance
async def handle(ctx):
    # Check if vision context is available
    if ctx.vision and ctx.vision.available:
        print(f"Auto-analyzed: {ctx.vision.description}")

    # Request raw frames for custom analysis
    frames = await ctx.request_frames(FrameRequestOptions(
        limit=5,
        raw_base64=True,
    ))

    if frames.frames:
        for frame in frames.frames:
            # frame.base64 contains the image data
            # frame.timestamp is when it was captured
            pass

    # Or get pre-analyzed descriptions
    analyzed = await ctx.request_frames(FrameRequestOptions(limit=3))
    if analyzed.descriptions:
        print(f"Frame descriptions: {analyzed.descriptions}")

    ctx.done("I can see what you're showing me!")

Using Memory

Agents with memory scope can query stored facts:

from voice_agent import MemoryQueryOptions

@agent.on_utterance
async def handle(ctx):
    # Query relevant memories
    memories = await ctx.query_memory(MemoryQueryOptions(
        query=ctx.text,
        top_k=5,
        threshold=0.7,
        types=["preference", "fact"],
    ))

    if memories.facts:
        context = "\n".join(f.content for f in memories.facts)
        response = await generate_with_context(ctx.text, context)
        ctx.done(response)
    else:
        ctx.done("I don't have any relevant memories about that.")

Handling Interrupts

When the user starts speaking, the gateway sends an interrupt:

@agent.on_utterance
async def handle(ctx):
    async for chunk in stream_response(ctx.text):
        # Check before each operation
        if ctx.is_aborted:
            print("User interrupted, stopping")
            return
        ctx.send_delta(chunk)
    ctx.done()

@agent.on_interrupt
def on_interrupt(session_id: str, reason: str):
    # reason: "new_user_speech" | "lost_arbitration" | "supersede"
    print(f"Session {session_id} interrupted: {reason}")

Configuration

from voice_agent import VoiceAgentConfig, ConnectionMode

config = VoiceAgentConfig(
    api_key="sk-voice-xxx",          # Your API key
    gateway_url="wss://...",          # Gateway WebSocket/HTTP URL
    mode=ConnectionMode.WEBSOCKET,    # WEBSOCKET (default) or SSE
    reconnect=True,                   # Auto-reconnect (default: True)
    reconnect_interval=1.0,           # Base reconnect delay in seconds
    max_reconnect_attempts=None,      # Max attempts (None = unlimited)
)

Context API

The UtteranceContext provides:

Property	Type	Description
text	str	The user's utterance text
is_final	bool	Whether this is a final transcript
user	UserInfo | None	User info (if profile/email/location scope)
vision	VisionContext | None	Vision context (if vision scope)
session_id	str	Current session ID
request_id	str	Current request ID
user_id	str | None	User ID
timestamp	datetime	When the utterance was received
is_aborted	bool	Whether the context was interrupted

Method	Description
send_delta(delta)	Stream a text chunk to the user
done(final_text=None)	Complete the response
request_frames(options=None)	Request video frames (async)
query_memory(options)	Query user memories (async)

Connection Modes

WebSocket (recommended)

Full-duplex communication, lower latency:

config = VoiceAgentConfig(
    mode=ConnectionMode.WEBSOCKET,
    # ...
)

Server-Sent Events (SSE)

One-way server push with HTTP POST for sending:

config = VoiceAgentConfig(
    mode=ConnectionMode.SSE,
    # ...
)

License

MIT