Per-User Assistant

Create isolated AI assistants for each user with persistent state

Build a system where each user gets their own AI assistant with isolated state, conversation history, and tools.

Quick Start

Use AgentServer to host multiple per-user agents:

from ai_query.agents import Agent, AgentServer, MemoryStorage
from ai_query.providers import openai

class UserAssistant(Agent):
    def __init__(self, id):
        super().__init__(
            id,
            model=openai("gpt-4o"),
            system="You are a personal assistant.",
            storage=MemoryStorage(),
            initial_state={"preferences": {}, "tasks": []}
        )

# Start server - handles routing to /agent/{user_id}/ws automatically
AgentServer(UserAssistant).serve(port=8080)

# Connect:
# ws://localhost:8080/agent/user-alice/ws
# ws://localhost:8080/agent/user-bob/ws

Complete Example: Personal Task Manager

from ai_query import tool, Field
from ai_query.agents import Agent, AgentServer, SQLiteStorage
from ai_query.providers import google
from datetime import datetime

class PersonalAssistant(Agent):
    def __init__(self, id):
        super().__init__(
            id,
            model=google("gemini-2.0-flash"),
            system="Help users manage tasks and stay organized.",
            storage=SQLiteStorage("./user_data.db"),
            initial_state={"tasks": []}
        )

    @tool(description="Add a new task")
    async def add_task(self, title: str = Field(description="Task title")) -> str:
        tasks = self.state.get("tasks", [])
        tasks.append({"title": title, "done": False})
        await self.update_state(tasks=tasks)
        return f"Added: {title}"

    async def on_connect(self, connection, ctx):
        await connection.send(f"Welcome back! You have {len(self.state['tasks'])} tasks.")

    async def on_message(self, connection, message):
        # Stream response
        async for chunk in self.stream(message):
            await connection.send(chunk)

if __name__ == "__main__":
    AgentServer(PersonalAssistant).serve(port=8080)

Authentication

Secure per-user access with middleware:

from ai_query.agents import AgentServer, AgentServerConfig

async def auth(request):
    token = request.headers.get("Authorization")
    return token == "secret-token"

config = AgentServerConfig(auth=auth)
AgentServer(PersonalAssistant, config=config).serve(port=8080)