hu-sdk 0.0.6
Python SDK for building voice agents on the Voice Gateway
Requires Python: >=3.10
Keywords: agent, assistant, llm, streaming, voice, websocket

# HU SDK (Python)

Python SDK for building voice agents on the Voice Gateway.

## Installation

```bash
pip install hu-sdk
```

## Quick Start

```python
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

```python
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:

```python
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:

```python
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:

```python
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:

```python
@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

```python
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) |
| `entities` | `list[EntityInfo]` | Entities extracted from the utterance (NER) |
| `topics` | `list[str]` | Topics detected in the utterance |
| `context` | `list[ContextUtterance]` | Previous utterances (if include_context filter set) |
| `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:

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

### Server-Sent Events (SSE)

One-way server push with HTTP POST for sending:

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

## License

MIT