Architecture Overview

The Big Picture

Agent.create(adapter=MyAdapter(), agent_id="...", api_key="...")
                            │
                            ▼
┌─────────────────────────────────────────────────────────┐
│                        Agent                            │
│            Composes: runtime + preprocessor + adapter   │
└─────────────────────────────────────────────────────────┘
       │                    │                    │
       ▼                    ▼                    ▼
┌──────────────-┐   ┌──────────────┐   ┌────────────────────┐
│PlatformRuntime│   │ Preprocessor │   │  SimpleAdapter[H]  │
│               │   │              │   │                    │
│• ThenvoiLink  │   │• Filters     │   │• HistoryConverter  │
│• AgentRuntime │   │  events      │   │• on_message()      │
└──────────────-┘   └──────────────┘   └────────────────────┘
Agent owns all three. PlatformRuntime owns ThenvoiLink + AgentRuntime.

Core Classes

Agent — Compositor

The top-level orchestrator. Doesn’t do work itself; coordinates three components.

1 agent = Agent.create(
2     adapter=MyAdapter(),
3     agent_id="...",
4     api_key="...",
5 )
6 await agent.run()

Owns	Purpose
`PlatformRuntime`	Platform connectivity
`Preprocessor`	Event filtering (runs in Agent’s event loop; returning `None` drops the event)
`FrameworkAdapter`	LLM framework logic

Method	Purpose
`run()`	Start + run forever + stop (typical usage)
`start()`	Manual: initialize runtime, call `adapter.on_started()`
`stop()`	Manual: shutdown runtime

SimpleAdapter[H] — Template Method

Generic base class that implements FrameworkAdapter protocol. H is your history type.

1 class MyAdapter(SimpleAdapter[list[ChatMessage]]):
2     def __init__(self):
3         super().__init__(history_converter=MyHistoryConverter())
4 
5     async def on_message(
6         self,
7         msg: PlatformMessage,
8         tools: AgentToolsProtocol,
9         history: list[ChatMessage],  # Fully typed!
10         participants_msg: str | None,
11         *,
12         is_session_bootstrap: bool,
13         room_id: str,
14     ) -> None:
15         # Your LLM logic here
16         ...

Method	When Called
`on_message()`	Each incoming message (abstract — you implement this)
`on_started()`	After platform connection
`on_cleanup()`	When leaving a room

History type depends on converter:

history_converter set → history is type H (converted)
history_converter is None → history is HistoryProvider (raw)

PlatformRuntime — Facade

Manages platform connectivity. Creates components lazily on start().

Creates	Purpose
`ThenvoiLink`	WebSocket + REST client
`AgentRuntime`	Room presence; maintains one `ExecutionContext` per room

Fetches agent metadata (name, description) before starting.

Protocols (Interfaces)

Protocol	Methods	Purpose
`FrameworkAdapter`	`on_event()`, `on_cleanup()`, `on_started()`	LLM framework contract
`AgentToolsProtocol`	`send_message()`, `execute_tool_call()`, `get_tool_schemas()`, …	Platform tools (pre-bound to `room_id` so LLM doesn’t need to know UUIDs)
`HistoryConverter[T]`	`convert(raw) → T`	History format conversion
`Preprocessor`	`process(ctx, event, agent_id) → AgentInput?`	Event filtering

All protocols are @runtime_checkable — duck typing with type safety.

Data Types

Type	Purpose	Key Fields
`PlatformMessage`	Immutable message	`id`, `content`, `sender_name`, `message_type`
`HistoryProvider`	Lazy history wrapper	`raw`, `convert(converter)`
`AgentInput`	Adapter input bundle	`msg`, `tools`, `history`, `is_session_bootstrap`
`PlatformEvent`	Tagged union	`MessageEvent \| RoomAddedEvent \| ...`

Data Flow

Inbound: Platform → Adapter

WebSocket
    → ThenvoiLink queues PlatformEvent
    → Preprocessor.process() filters + creates AgentInput
    → Adapter.on_message(msg, tools, history, ...)

Outbound: Adapter → Platform

Pattern 2 (adapter manages tool loop):

LLM returns tool_calls
    → tools.execute_tool_call(name, args)
    → AgentTools dispatches to REST API
    → Platform receives action

Pattern 1 (framework manages tools): The framework executes tools internally; adapter just forwards streaming events to the platform via tools.send_event().

Package Layout

thenvoi/
├── agent.py              # Agent compositor
├── core/
│   ├── protocols.py      # FrameworkAdapter, AgentToolsProtocol, etc.
│   ├── types.py          # PlatformMessage, AgentInput, HistoryProvider
│   └── simple_adapter.py # SimpleAdapter[H] base class
├── adapters/             # LangGraph, Anthropic, PydanticAI, ClaudeSDK
├── converters/           # History converters per framework
├── platform/
│   ├── link.py           # ThenvoiLink (WebSocket + REST)
│   └── event.py          # PlatformEvent tagged union
├── runtime/
│   ├── tools.py          # AgentTools implementation
│   ├── execution.py      # ExecutionContext (per-room state)
│   └── ...
└── testing/
    └── fake_tools.py     # FakeAgentTools for unit tests

Extension Points

Want to…	Extend/Implement
Add new LLM framework	`SimpleAdapter[H]` + `HistoryConverter[H]`
Custom event filtering	`Preprocessor` protocol
Mock tools in tests	Use `FakeAgentTools`

Design Patterns

Pattern	Where Used
Composition over Inheritance	Agent composes runtime, adapter, preprocessor
Protocol-Based Contracts	All interfaces are protocols (duck typing)
Generic Type Parameters	`SimpleAdapter[H]`, `HistoryConverter[T]`
Tagged Union	`PlatformEvent` for type-safe event matching
Lazy Initialization	PlatformRuntime creates components on `start()`
Strategy Pattern	HistoryConverter swappable at runtime

Concurrency Model

Gotcha for adapter authors

on_message() is called sequentially per room (messages in a room are processed one at a time)
Multiple rooms run concurrently (each room has its own asyncio task)
Do not share mutable state across rooms without synchronization (e.g., use dict[room_id, state] not a global variable)