HU SDK (Python)

Python SDK for building voice agents on the Voice Gateway.

Installation

pip install hu-sdk

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.")

Routing Filters

Agents can register filters to control which utterances are routed to them. Filters are evaluated server-side for efficient routing in multi-agent setups:

await agent.connect()

# Register filters after connecting
agent.register_filters(
    # Match utterances containing these entity types or values
    entities=["PERSON", "John"],
    # Match utterances about these topics
    topics=["weather", "travel"],
    # Match utterances containing these keywords
    keywords=["urgent", "help"],
    # Match specific speakers
    speakers=["user"],
    # Number of previous utterances to include for context (used with "filtered" tier)
    include_context=5,
    # Data access tier - controls what data the agent receives:
    # - "full": everything (whole conversation stream)
    # - "filtered": matching messages + context window (default)
    # - "summary": just {entities, topics} - no text
    tier="filtered",
)

Filters can be updated at any time while connected. The gateway will apply the new filters to subsequent utterances.

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:

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