OpenBotX Architecture

OpenBotX Architecture

OpenBotX is an AI assistant platform. It provides the infrastructure for running autonomous AI agents that communicate through multiple channels, execute tools, manage tasks, and maintain persistent memory -- all orchestrated through an async message bus.

This document describes the system architecture, core components, and data flow.

Table of Contents

• High-Level Overview
• Message Flow
• Real-Time Updates
• Core Components
  • Server
  • Message Bus
  • Agent
  • Channels
  • Credentials
  • Providers
  • Tools
  • Tasks
  • Sessions
  • Cron
  • Heartbeat
  • Config
  • Helpers
  • Storage
  • Web Client
• Package Structure
• Startup Lifecycle
• Key Design Decisions

High-Level Overview

OpenBotX is composed of the following layers:

1. Channels -- Ingest user messages from the web UI, Telegram, or other integrations.
2. Message Bus -- Async queue pair (inbound / outbound) that decouples channels from the agent.
3. Orchestrator -- Consumes inbound messages, classifies them to the appropriate agent (when multiple agents are configured), and delegates processing.
4. Agent Loop -- Runs an agentic LLM loop (call model, execute tools, repeat), and publishes the final response to the outbound queue.
5. Channel Manager -- Consumes outbound messages and routes them back to the originating channel.
6. Event Dispatcher -- Broadcasts real-time events (thinking, tool use, messages, task updates) through registered handlers (e.g. WebSocketManager). Decouples event producers from the transport layer.

Supporting services include task management, session persistence, memory consolidation, scheduled jobs (cron), and a configurable tool registry.

Message Flow

The primary request/response path:

User
  --> Channel (Web / Telegram)
    --> MessageBus (inbound queue)
      --> Orchestrator
        --> AgentClassifier (selects agent, if multi-agent)
        --> AgentLoop
          --> LLM Provider (chat completion)
            --> Tool execution (if tool calls present)
              --> (repeat until no more tool calls or max iterations)
      --> MessageBus (outbound queue)
        --> ChannelManager
          --> Channel
            --> User

Each inbound message creates a Task object that tracks the request through its lifecycle: TODO -> DOING -> DONE (or ERROR).

Real-Time Updates

Browser clients connect via WebSocket at /ws (authenticated with a JWT token in the query string). The WebSocketManager broadcasts events to all connected clients as the agent works:

AgentLoop --> WebSocketManager --> Browser

Event types:

The WebSocket endpoint also accepts chat:send messages from the browser, which are converted to InboundMessage objects and published to the message bus.

Core Components

Server

Location: openbotx/server/

The server is a FastAPI application with a lifespan context manager that initializes and tears down all services. The ServerFactory class encapsulates all dependency creation logic.

ServerFactory encapsulates all dependency creation:

The built web client (openbotx/web_client/) is served as a SPA at /app/ with a catch-all fallback to index.html. The build output lives inside the Python package and is included in the .whl distribution.

Files under the project's public/ directory are served at /public/{path} without authentication. This allows media files (images, video, audio) to be embedded in HTML5 tags (<img>, <video>, <audio>) without needing auth headers.

Message Bus

Location: openbotx/bus/

The MessageBus is the backbone of the platform. It holds two asyncio.Queue instances that fully decouple message producers (channels) from the consumer (orchestrator/agent loop).

Session key derivation: Each InboundMessage derives its session key as {channel}:{chat_id}, which ties all messages from the same chat to the same conversation session. This can be overridden via session_key_override.

Agent

Location: openbotx/agent/

The agent subsystem is the intelligence layer of the platform.

Multi-agent orchestration:

When multiple agents are defined in the config, the Orchestrator uses the AgentClassifier to determine which agent should handle each message:

1. Orchestrator receives InboundMessage from bus (1-second timeout polling for graceful shutdown)
2. If single agent → route to default agent (no classification overhead)
3. If multiple agents → call AgentClassifier:
   a. Build system prompt listing all agents and their descriptions
   b. Prepare history: last 20 messages, assistant messages prefixed with [Agent: name]
   c. Send to LLM (max_tokens=256, temperature=0.0)
   d. LLM calls `route(agent_name, confidence)` tool
   e. Validate agent_name exists in configured agents
   f. Return agent_name (or default on unknown/error)
4. Delegate message processing to selected AgentLoop

The classifier uses the model specified in classifier.model (config), falling back to the default agent's model if not set.

Classifier system prompt rules:

The classifier operates with four rules:

1. Analyze the user's latest message and the conversation history.
2. Continuity bias: If the conversation was previously handled by a specific agent, continue with that agent unless the topic clearly changes. This prevents unnecessary agent switches mid-conversation.
3. Use the route tool to select the best agent.
4. Always select exactly one agent from the available list.

Assistant messages in the classifier's history include an [Agent: name] prefix (e.g., [Agent: crypto] Here's the market data...), so the classifier can see which agent handled previous turns and maintain continuity.

Inter-agent communication:

Agents do not communicate directly with each other. There is no message passing, shared queue, or RPC mechanism between agents. Each inbound message is routed to exactly one agent by the classifier, and only that agent processes it.

The only form of "communication" is indirect through the shared session history. All agents in the same chat use the same session (keyed by {channel}:{chat_id}). When the classifier switches from one agent to another, the new agent sees the full conversation history, including everything the previous agent said.

Example with two agents (crypto and assistant):

1. User: "What is the current price of Bitcoin?"
   → Classifier selects "crypto" (matches topic)
   → crypto agent responds with market data
   → Response saved to session history

2. User: "Thanks, now help me write an email"
   → Classifier detects topic change, selects "assistant"
   → assistant agent receives the full session history,
     including the crypto agent's previous response
   → assistant agent responds about the email
   → Response saved to session history

3. User: "Compare that price with Ethereum"
   → Classifier sees crypto context returning, selects "crypto"
   → crypto agent sees the entire conversation (its own earlier
     response + the assistant's email response + the new message)
   → crypto agent responds with the comparison

Each agent is unaware that other agents exist. They just see a conversation history with user and assistant messages. The classifier is the only component that knows about all agents and decides which one handles each turn.

Orchestrator error resilience:

The orchestrator's run() loop catches exceptions per-message. If a single message fails (e.g., the agent loop throws), the error is logged but the orchestrator continues processing the next message. This prevents one bad request from crashing the entire system. The 1-second timeout on consume_inbound() allows the stop event to be checked regularly for graceful shutdown.

Agentic loop detail:

1. Receive InboundMessage from Orchestrator
2. Create or resume Task (set state to DOING)
3. Initialize task.live_state = {tool_uses: []}
4. Load or create Session
5. Initialize session.live_state = {tool_uses: [], agent_name: ...}
6. Build system prompt (ContextBuilder)
7. Build message array (system + history + new user message)
8. Loop:
   a. Increment task.iteration_count
   b. Call LLM provider with messages + tool definitions
   c. If response contains tool_calls:
      - Execute each tool via ToolRegistry
      - Increment task.tool_count per tool call
      - Broadcast chat:tool_use via EventDispatcher
      - Append tool entry to session.live_state and task.live_state
      - Append assistant message + tool results to messages
      - Continue loop
   d. If response is plain text:
      - Break loop, return response
9. Clear session.live_state and task.live_state
10. Save messages to session
11. Publish OutboundMessage to bus
12. Set Task state to DONE
13. Check if memory consolidation is needed

Channels

Location: openbotx/channels/

Channels are the communication endpoints that connect users to the platform.

The web channel is implicit and does not have a BaseChannel implementation. The WebSocket endpoint handles web client communication directly. ChannelManager._route_message broadcasts web-bound outbound messages via WebSocketManager.

Credentials

Credentials centralize all secrets in one place. Each credential has a type (e.g., simple, oauth1, basic, login, aws) and type-specific fields. Other configuration sections reference credentials by name via their credential field, rather than storing secrets directly.

The CredentialConfig model serializes only the fields relevant to its type, keeping the stored configuration clean.

Providers

Location: openbotx/providers/

The provider subsystem abstracts LLM access behind a uniform interface.

Provider resolution: The Config.get_provider() method matches a model name to a provider by first checking the LiteLLM-style prefix (e.g., anthropic/claude-sonnet-4-20250514), then falling back to keyword matching, and finally returning any provider whose referenced credential has an API key configured. ServerFactory.create_provider() resolves through provider -> credential -> concrete key to build a LiteLLMProvider.

Provider-level model_params: Each ProviderConfig can define a default model_params dict (arbitrary key-value pairs like max_tokens, temperature, top_p, etc.). At startup, ServerFactory.create_orchestrator merges provider defaults into each agent's model_params using simple dict merge ({**provider_params, **agent_params}). Agent-level keys always take precedence.

Tools

Location: openbotx/tools/

Tools are the actions the agent can perform in the world.

Subagent tool restrictions: When SubagentManager builds a tool registry for a subagent, it includes file operations, shell, web tools, HTTP client, RSS reader, browser, and image generation. It excludes MessageTool, SpawnTool, CronTool, and all memory tools to prevent subagents from sending messages to users, spawning further subagents, creating scheduled jobs, or modifying memory.

Tasks

Location: openbotx/tasks/

Tasks provide observability into what the agent is doing.

Task states follow a Kanban model:

Tasks support a parent-child relationship: subagent tasks reference their parent_task_id, enabling hierarchical task tracking.

Retention: The GET /api/tasks endpoint excludes DONE and ERROR tasks older than 24 hours, keeping the task board focused on recent activity.

Sessions

Location: openbotx/session/

Sessions persist conversation history.

The Session dataclass holds a list of messages (role + content + metadata) and tracks last_consolidated for memory consolidation. get_history() returns messages for building LLM context, capped at 500 by default. The transient live_state dict holds runtime data (e.g., tool_uses, agent_name) during execution. It is returned via the chat history API but never persisted to JSONL.

Cron

Location: openbotx/cron/

The cron service enables scheduled task execution.

Schedule kinds:

Jobs marked delete_after_run: true are automatically removed after their first execution.

Heartbeat

Location: openbotx/heartbeat/

The heartbeat service periodically checks HEARTBEAT.md in the workspace for tasks.

Unlike cron (agent-managed via tools), HEARTBEAT.md is a file the user edits manually — a persistent to-do list the agent checks periodically.

Config

Location: openbotx/config/

Configuration is defined as Pydantic models and loaded from YAML.

Key configuration sections:

AgentConfig includes:

• resolve_workspace(project_path) method that resolves the workspace path relative to the project root.
• @field_validator("workspace") that defaults empty or null values to "./workspace".
• description field used by the AgentClassifier for routing decisions.
• instructions field appended to the system prompt as agent-specific instructions.
• tools list that whitelists which tools are available to the agent.

Helpers

Location: openbotx/helpers/

Utility modules shared across the codebase.

Storage

Location: openbotx/storage/

Pluggable storage backends for workspace files. The StorageProvider abstraction supports both file and directory operations, allowing the Files API and other components to work uniformly across all backends.

Web Client

Location: web_client/

The web client is a single-page application built with:

• Vue 3 for components
• Vite for build and dev server
• PrimeVue 4 for UI components
• Tailwind CSS 4 for styling
• Pinia for state management
• md-editor-v3 for Markdown editing and preview
• WebSocket for real-time communication

Pages:

Package Structure

openbotx/
├── agent/           # AI agent loop, orchestration, classification, context, memory, skills, subagents
│   ├── orchestrator.py    # Orchestrator - message routing to agents
│   ├── classifier.py      # AgentClassifier - LLM-based agent selection
│   ├── loop.py            # AgentLoop - main agentic processing loop
│   ├── context.py         # ContextBuilder - system prompt assembly
│   ├── memory.py          # MemoryStore - conversation memory persistence
│   ├── skills.py          # SkillsLoader - SKILL.md discovery and loading
│   └── subagent.py        # SubagentManager - background task delegation
├── cdp/             # Vendored Chrome DevTools Protocol library (from python-cdp)
│   ├── base.py          # IEventLoop protocol
│   ├── exceptions.py    # CDP exception classes
│   ├── utils.py         # LoggerMixin, Retry, Worker utilities
│   ├── browser.py       # ChromeLauncher - Chrome process management
│   ├── connection.py    # CDPConnection, CDPSession, connect_cdp (WebSocket)
│   └── protocol/        # Auto-generated CDP domain modules (runtime, page, dom, input_, target, etc.)
├── bus/             # Async message bus and event dispatching
│   ├── queue.py         # MessageBus with inbound/outbound queues
│   ├── events.py        # InboundMessage, OutboundMessage data classes
│   └── dispatcher.py    # EventDispatcher - protocol-based event broadcasting
├── channels/        # Communication channel implementations
│   ├── base.py          # BaseChannel abstract interface
│   ├── manager.py       # ChannelManager - lifecycle and routing
│   └── telegram.py      # TelegramChannel integration
├── cli/             # CLI commands (init, start, version)
├── config/          # Configuration
│   ├── schema.py        # Pydantic configuration models
│   └── loader.py        # YAML loader with env var expansion
├── cron/            # Scheduled task service
│   ├── service.py       # CronService - tick loop and job execution
│   └── types.py         # CronJob, CronSchedule, CronPayload data classes
├── heartbeat/       # Periodic HEARTBEAT.md checker
│   └── service.py       # HeartbeatService - reads workspace/HEARTBEAT.md
├── helpers/         # Utility modules
│   ├── path.py          # PathResolver - workspace-scoped path resolution and directory restrictions
│   ├── oauth1.py        # OAuth 1.0a signature generation (HMAC-SHA1)
│   ├── ssrf.py          # SSRF protection - blocks requests to private/internal networks
│   ├── transcription.py # Audio transcription via faster-whisper
│   ├── text.py          # Text formatting utilities (humanize, describe_tool_use)
│   └── secrets.py       # Sensitive value masking for config display
├── providers/       # LLM model provider abstraction
│   ├── base.py          # LLMProvider and LLMResponse
│   ├── litellm_provider.py  # LiteLLM wrapper for multi-provider access
│   └── registry.py      # ProviderSpec definitions and matching
├── server/          # FastAPI server
│   ├── app.py           # ServerFactory, lifespan, and create_app
│   ├── websocket.py     # WebSocketManager and endpoint
│   ├── auth.py          # JWT authentication middleware
│   └── routes/          # REST API endpoint routers
├── session/         # Conversation session management
│   └── manager.py       # SessionManager with JSONL persistence
├── skills/          # Built-in skill definitions (SKILL.md files)
├── storage/         # Storage backends
│   ├── base.py          # Abstract storage interface
│   ├── local.py         # Local filesystem storage
│   └── s3.py            # AWS S3 storage
├── tasks/           # Task management
│   ├── models.py        # Task model and TaskState enum
│   └── manager.py       # TaskManager with event broadcasting
├── tools/           # Built-in tool implementations
│   ├── base.py          # Abstract Tool class
│   ├── registry.py      # ToolRegistry - registration and execution
│   ├── filesystem.py    # read_file, write_file, edit_file, list_dir (PathResolver)
│   ├── shell.py         # exec (shell command execution)
│   ├── web.py           # web_search, web_fetch
│   ├── message.py       # message (send to channels)
│   ├── spawn.py         # spawn (delegate to subagent)
│   ├── cron.py          # cron (manage scheduled jobs)
│   ├── memory_tool.py   # memory_save, memory_read, memory_search
│   ├── browser.py       # browser (CDP-based browser automation)
│   ├── http_client.py   # http_client (HTTP requests with auth profiles)
│   ├── rss.py           # rss_reader (RSS/Atom feed reader)
│   └── image.py         # generate_image (AI image generation)
└── version.py       # Package version

Startup Lifecycle

The FastAPI lifespan context manager in app.py orchestrates the full startup sequence. The ServerFactory class handles all dependency creation.

1. Load configuration from YAML
2. Create ServerFactory from config
3. Ensure system workspace directory exists
4. Setup logging (rotating file handler + console)
5. Auto-create server.credential (simple, random key) if not configured; auto-create web_client credential (login, admin/admin) if

…(truncated)