CHAPTER 1: THE VISION AND ARCHITECTURE
Welcome to one of the most exciting frontiers in software engineering today. Agentic AI is not simply a chatbot that answers questions. It is a system where one or more AI-driven agents pursue goals autonomously, reason about what steps to take, call tools when needed, reflect on results, and iterate until a goal is achieved. When you combine multiple such agents running concurrently, each with their own conversation history, their own chosen LLM model, and their own set of tools, you have something genuinely powerful.
This tutorial builds exactly that system from the ground up. You will end up with a production-quality application with the following characteristics.
The application presents itself as a main chat window. This main window is always the chatbot itself, meaning the application never navigates away from it. Each new prompt the user starts opens as a new tab inside that window. Multiple tabs can be active simultaneously, each running its own agent or simple chat session concurrently. This is the concurrent orchestration: the application manages all these running sessions in parallel, coordinating their LLM calls, tool executions, and state updates without any of them blocking the others.
Each tab maintains its own session history, meaning the full conversation thread for that particular interaction is kept alive and visible while the tab is open. When the user closes a tab, the session is automatically saved to a sessions list. From this sessions list, the user can later select any completed session and review the full conversation history. Deleting all sessions removes both active and saved chats, giving the user a clean slate.
The model management subsystem allows users to add LLM models by providing a display name, a base URL, and an authentication key. This design supports both local models (such as those served by Ollama or LM Studio on localhost) and remote models (such as OpenAI, Mistral, or any OpenAI-compatible API endpoint). For each model, the application tracks how many input tokens and output tokens the user has consumed, displaying this information directly in the model list. A default model is always defined so that new prompts work immediately without requiring the user to configure anything.
Each tab can be toggled into agent mode by checking a checkbox. In agent mode, the user's prompt is treated as a goal rather than a simple question. The agent then uses the ReACT pattern (Reasoning, Acting, Observing) to pursue that goal: it reasons about what to do, selects a tool to call, observes the result, and repeats until it concludes. MCP (Model Context Protocol) servers can be added to the application, and agents automatically discover and use the tools those servers expose.
THE LAYERED ARCHITECTURE
The application follows a clean layered architecture. At the bottom is the data layer, which handles persistence using SQLite with async access. Above that is the service layer, which contains the business logic for LLM communication, agent execution, session management, and MCP tool calling. Above the service layer is the orchestration layer, which manages concurrent agent execution using Python's asyncio. At the top is the API layer, built with FastAPI, which exposes WebSocket endpoints for real-time streaming and REST endpoints for configuration management. The frontend is a single-page web application that communicates with the backend over WebSockets and REST.
The directory structure of the complete application is:
agentic_chatbot/
backend/
__init__.py
main.py
config.py
database.py
orchestrator.py
models/
__init__.py
llm_model.py
session.py
agent.py
mcp_server.py
services/
__init__.py
llm_service.py
agent_service.py
mcp_service.py
session_service.py
frontend/
index.html
app.js
style.css
requirements.txt
This structure separates concerns cleanly. The models directory contains pure data definitions with no business logic. The services directory contains all business logic with no knowledge of HTTP or WebSockets. The orchestrator coordinates concurrent execution. The main.py file wires everything together and exposes the API.
CHAPTER 2: TECHNOLOGY STACK AND PROJECT SETUP
The technology choices here are deliberate and worth explaining in detail.
Python 3.11 or higher is used for the backend because its asyncio implementation is mature, its type annotation system is excellent, and the AI/ML ecosystem is richest in Python. FastAPI is chosen as the web framework because it has first-class support for async/await, WebSockets, and automatic OpenAPI documentation. It is also extremely fast and easy to understand.
For LLM communication, the application uses the httpx library for async HTTP calls. This allows streaming responses from LLMs without blocking. The design uses the OpenAI Chat Completions API format as a universal interface, because virtually every modern LLM server (Ollama, LM Studio, vLLM, OpenAI, Mistral, Groq, and many others) either natively speaks this format or provides a compatibility layer for it.
SQLite with the aiosqlite library provides async-safe persistence without requiring a separate database server. This keeps the application self-contained and easy to deploy.
For MCP (Model Context Protocol) integration, the application uses FastMCP 3, the official high-level Python client library for MCP. FastMCP 3 provides a unified Client class that handles both stdio-based and HTTP-based server connections, manages the full protocol handshake automatically, and exposes a clean async API for tool discovery and invocation. This replaces the need for any manual JSON-RPC subprocess management.
The frontend uses vanilla JavaScript with no framework dependencies. This is a deliberate choice to keep the tutorial accessible and to demonstrate that a sophisticated UI does not require a heavy framework.
Start by creating the project and installing dependencies:
mkdir agentic_chatbot
cd agentic_chatbot
python -m venv venv
source venv/bin/activate
# On Windows use: venv\Scripts\activate
The requirements.txt file lists all dependencies:
fastapi>=0.116.0
uvicorn[standard]>=0.32.0
httpx>=0.28.0
aiosqlite>=0.21.0
pydantic>=2.9.0
python-multipart>=0.0.12
websockets>=14.0
aiofiles>=24.1.0
fastmcp>=3.0.0
Install them with:
pip install -r requirements.txt
Now create the directory structure:
mkdir -p backend/models backend/services frontend
The three init.py files that make the directories into Python packages are empty files. Create them with:
touch backend/__init__.py
touch backend/models/__init__.py
touch backend/services/__init__.py
On Windows, use:
type nul > backend\__init__.py
type nul > backend\models\__init__.py
type nul > backend\services\__init__.py
Each __init__.py file is intentionally empty. Its sole purpose is to tell Python that the directory is a package, enabling the import system to find modules within it.
CHAPTER 3: DATA MODELS AND CLEAN ARCHITECTURE
Good architecture starts with clear data models. These models define the vocabulary of the entire application. Every service, every API endpoint, and every piece of frontend logic speaks in terms of these models. Getting them right at the start saves enormous amounts of refactoring later.
The configuration file establishes application-wide settings. Keeping all configuration in one place means you never have to hunt through multiple files to change a timeout value or a database path.
# backend/config.py
"""
Application-wide configuration.
All tuneable parameters live here so that changing the behaviour of
the application never requires touching business-logic files.
"""
from pydantic import BaseModel
class AppConfig(BaseModel):
"""
Central configuration for the Agentic Chatbot application.
Pydantic validates the types of all fields at construction time,
catching misconfiguration errors immediately at startup rather
than at runtime when a feature is first used.
"""
app_name: str = "Agentic Chatbot Orchestrator"
version: str = "1.0.0"
database_path: str = "chatbot.db"
default_model_id: str = "default"
max_agent_iterations: int = 15
agent_timeout_seconds: float = 300.0
stream_chunk_size: int = 64
cors_origins: list[str] = ["*"]
# Single global config instance used throughout the application.
config = AppConfig()
The LLM model data model represents a configured language model. Notice that it carries token usage statistics directly on the model record. This means every time the application makes an LLM call, it updates these counters on the model record and persists them. The user sees live, accurate token consumption per model.
# backend/models/llm_model.py
"""
Data model for a configured LLM endpoint.
This model represents everything the application needs to know about
a language model: where to reach it, how to authenticate, and how
much it has been used. Both local models (Ollama, LM Studio) and
remote models (OpenAI, Mistral, etc.) are represented identically
here, because they all speak the OpenAI Chat Completions API format.
"""
import uuid
from typing import Optional
from pydantic import BaseModel, Field
class LLMModel(BaseModel):
"""
Represents a configured LLM endpoint with usage tracking.
The base_url field points to the root of an OpenAI-compatible API.
For Ollama running locally this would be 'http://localhost:11434'.
For OpenAI this would be 'https://api.openai.com'.
For LM Studio this would be 'http://localhost:1234'.
The model_name field is the specific model identifier sent in API
requests, such as 'llama4', 'gpt-4o', or 'mistral-nemo'.
Token counters accumulate over the lifetime of the model record
and are updated after every successful LLM call.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
display_name: str
base_url: str
model_name: str
api_key: str = ""
is_default: bool = False
input_tokens_used: int = 0
output_tokens_used: int = 0
created_at: str = ""
The session model represents a complete conversation, whether it is still active in a tab or has been saved to the sessions list. The ToolCall model captures every tool invocation made by an agent, including the arguments passed and the result returned.
# backend/models/session.py
"""
Data models for conversation sessions and individual messages.
A Session is the top-level container for a conversation. It holds
metadata about the conversation and a list of messages. When a tab
is closed, its Session is serialized and saved to the database.
A Message represents a single turn in the conversation. The role
field follows the OpenAI convention: 'system', 'user', 'assistant',
or 'tool'. The tool_calls field supports the ReACT agent pattern
where the assistant calls tools and observes results.
"""
import uuid
from typing import Optional, Any
from pydantic import BaseModel, Field
class ToolCall(BaseModel):
"""
Represents a single tool invocation made by the agent.
Stores the tool name, the arguments passed, and the result
returned by the MCP server. The tool_call_id links the call
to its result in the conversation history, which is required
by the OpenAI function-calling protocol.
"""
tool_call_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
tool_name: str
arguments: dict[str, Any] = {}
result: Optional[str] = None
error: Optional[str] = None
class Message(BaseModel):
"""
A single message in a conversation.
The content field holds the text of the message. For assistant
messages that include tool calls, the tool_calls list is populated.
For tool result messages, the role is 'tool' and the content
contains the tool output.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
role: str # 'system' | 'user' | 'assistant' | 'tool'
content: str
tool_calls: list[ToolCall] = []
timestamp: str = ""
thinking: Optional[str] = None # ReACT reasoning trace
class Session(BaseModel):
"""
A complete conversation session, corresponding to one tab.
The tab_id links this session to its active tab in the frontend.
When the tab is closed, tab_id is cleared and the session moves
to the saved sessions list.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
title: str = "New Chat"
tab_id: Optional[str] = None
model_id: str = ""
is_agent_mode: bool = False
messages: list[Message] = []
is_active: bool = True
created_at: str = ""
updated_at: str = ""
The agent model defines a named agent configuration. When the user enables agent mode in a tab, the tab uses an agent definition that specifies which MCP servers it can access and other behavioural parameters.
# backend/models/agent.py
"""
Data model for agent configuration.
An agent is a named configuration that extends a chat session with
autonomous goal-pursuit capabilities. The agent uses the ReACT pattern
and has access to tools provided by configured MCP servers.
"""
import uuid
from pydantic import BaseModel, Field
# The default system prompt instills the ReACT mindset into the agent.
# It instructs the model to think step by step, use tools when needed,
# and always deliver a clear final answer to the user.
DEFAULT_AGENT_SYSTEM_PROMPT = (
"You are a helpful AI agent. You have access to tools that you "
"can use to accomplish the user's goals. Think step by step, "
"use tools when needed, and always provide a clear final answer."
)
class AgentConfig(BaseModel):
"""
Configuration for a ReACT agent.
The system_prompt field provides the agent's core instructions.
The mcp_server_ids list specifies which MCP servers this agent
has access to. The max_iterations field caps the ReACT loop to
prevent runaway agents consuming unlimited tokens.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
name: str
description: str = ""
system_prompt: str = DEFAULT_AGENT_SYSTEM_PROMPT
mcp_server_ids: list[str] = []
max_iterations: int = 15
created_at: str = ""
The MCP server model represents a configured Model Context Protocol server. FastMCP 3 supports both stdio (for local process-based servers) and HTTP (for remote or containerised servers) transports.
# backend/models/mcp_server.py
"""
Data model for MCP (Model Context Protocol) server configuration.
FastMCP 3 supports two transport mechanisms: stdio (for local
process-based servers) and HTTP (for remote or containerized servers).
This model captures the configuration needed for both transport types.
"""
import uuid
from typing import Optional, Literal
from pydantic import BaseModel, Field
class MCPServer(BaseModel):
"""
Configuration for an MCP tool server.
For stdio transport, the command field specifies the executable
and args provides its command-line arguments. This is used for
local MCP servers such as filesystem, git, or database tools.
For http transport, the url field specifies the endpoint of the
remote MCP server. An optional api_key can be provided for
authentication.
The cached_tools field stores the list of tool schemas discovered
from this server in OpenAI function-calling format. Caching avoids
repeated round-trips to the server on every agent iteration.
"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
name: str
transport: Literal["stdio", "http"] = "stdio"
command: Optional[str] = None # For stdio transport
args: list[str] = [] # For stdio transport
url: Optional[str] = None # For http transport
api_key: Optional[str] = None # For http transport
env: dict[str, str] = {} # Environment variables
is_active: bool = True
cached_tools: list[dict] = [] # OpenAI-format tool schemas
created_at: str = ""
CHAPTER 4: THE DATABASE LAYER
The database layer uses SQLite with aiosqlite for fully async operation. The design uses a simple but effective schema where complex nested objects like message lists are stored as JSON. This avoids the complexity of a fully normalised relational schema while still providing persistence, queryability, and reasonable performance for the use case.
The schema is created using individual execute calls rather than executescript. This gives finer control over error handling and avoids the implicit COMMIT that executescript issues before running, which can interfere with explicit transaction management.
The update_token_usage method uses SQL addition (column + ?) rather than reading the current value, adding to it in Python, and writing it back. This is critical for correctness when multiple agents are running concurrently and both happen to be using the same LLM model. If two agents both read the value as 1000, both add 50, and both write 1050, you lose 50 tokens from the count. The SQL approach makes the increment atomic at the database level.
# backend/database.py
"""
Async SQLite database layer for the Agentic Chatbot.
This module handles all database operations using aiosqlite, which
provides an async wrapper around Python's built-in sqlite3 module.
All methods are async and safe to call from asyncio coroutines.
The schema stores LLM models, sessions, agent configurations, and
MCP server configurations. Complex nested structures like message
lists are stored as JSON strings, which keeps the schema simple
while preserving full fidelity of the data.
"""
import json
import logging
from datetime import datetime, timezone
from typing import Optional
import aiosqlite
from backend.config import config
from backend.models.agent import AgentConfig
from backend.models.llm_model import LLMModel
from backend.models.mcp_server import MCPServer
from backend.models.session import Message, Session, ToolCall
logger = logging.getLogger(__name__)
def _now_iso() -> str:
"""Return the current UTC time as an ISO 8601 string."""
return datetime.now(timezone.utc).isoformat()
class Database:
"""
Manages all SQLite interactions for the application.
A single Database instance is created at startup and shared
across all services. The connection is opened once and reused,
which is safe with aiosqlite's async locking.
"""
def __init__(self, db_path: str) -> None:
self.db_path = db_path
self._conn: Optional[aiosqlite.Connection] = None
async def connect(self) -> None:
"""
Open the database connection and initialise the schema.
Called once at application startup.
"""
self._conn = await aiosqlite.connect(self.db_path)
# WAL mode allows concurrent reads while a write is in progress,
# which is important for an application with many concurrent tasks.
await self._conn.execute("PRAGMA journal_mode=WAL")
await self._conn.execute("PRAGMA foreign_keys=ON")
await self._create_schema()
await self._seed_default_model()
logger.info("Database connected: %s", self.db_path)
async def disconnect(self) -> None:
"""Close the database connection gracefully."""
if self._conn:
await self._conn.close()
logger.info("Database disconnected.")
async def _create_schema(self) -> None:
"""
Create all tables if they do not already exist.
Using IF NOT EXISTS makes this idempotent: it is safe to call
on every startup without wiping existing data. Each table is
created in a separate execute call so that a failure in one
statement does not silently skip the remaining statements.
"""
statements = [
"""
CREATE TABLE IF NOT EXISTS llm_models (
id TEXT PRIMARY KEY,
display_name TEXT NOT NULL,
base_url TEXT NOT NULL,
model_name TEXT NOT NULL,
api_key TEXT DEFAULT '',
is_default INTEGER DEFAULT 0,
input_tokens_used INTEGER DEFAULT 0,
output_tokens_used INTEGER DEFAULT 0,
created_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
model_id TEXT NOT NULL,
is_agent_mode INTEGER DEFAULT 0,
messages TEXT DEFAULT '[]',
is_active INTEGER DEFAULT 1,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS agent_configs (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
description TEXT DEFAULT '',
system_prompt TEXT NOT NULL,
mcp_server_ids TEXT DEFAULT '[]',
max_iterations INTEGER DEFAULT 15,
created_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS mcp_servers (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
transport TEXT NOT NULL,
command TEXT,
args TEXT DEFAULT '[]',
url TEXT,
api_key TEXT,
env TEXT DEFAULT '{}',
is_active INTEGER DEFAULT 1,
cached_tools TEXT DEFAULT '[]',
created_at TEXT NOT NULL
)
""",
]
for statement in statements:
await self._conn.execute(statement)
await self._conn.commit()
async def _seed_default_model(self) -> None:
"""
Ensure a default LLM model exists in the database.
This seeds Ollama running on localhost with llama4 as the
default, which works out of the box for local development.
Users can change the default model through the UI at any time.
"""
cursor = await self._conn.execute(
"SELECT COUNT(*) FROM llm_models WHERE is_default = 1"
)
row = await cursor.fetchone()
if row[0] == 0:
default_model = LLMModel(
id=config.default_model_id,
display_name="Local Ollama (llama4)",
base_url="http://localhost:11434",
model_name="llama4",
api_key="",
is_default=True,
created_at=_now_iso(),
)
await self.save_llm_model(default_model)
logger.info("Seeded default LLM model.")
# ------------------------------------------------------------------
# LLM Model CRUD
# ------------------------------------------------------------------
async def save_llm_model(self, model: LLMModel) -> None:
"""Insert or replace an LLM model record."""
await self._conn.execute(
"""
INSERT OR REPLACE INTO llm_models
(id, display_name, base_url, model_name, api_key,
is_default, input_tokens_used, output_tokens_used,
created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
model.id,
model.display_name,
model.base_url,
model.model_name,
model.api_key,
1 if model.is_default else 0,
model.input_tokens_used,
model.output_tokens_used,
model.created_at or _now_iso(),
),
)
await self._conn.commit()
async def get_all_llm_models(self) -> list[LLMModel]:
"""Retrieve all configured LLM models, default first."""
cursor = await self._conn.execute(
"SELECT * FROM llm_models "
"ORDER BY is_default DESC, display_name ASC"
)
rows = await cursor.fetchall()
return [self._row_to_llm_model(r) for r in rows]
async def get_llm_model(self, model_id: str) -> Optional[LLMModel]:
"""Retrieve a single LLM model by ID."""
cursor = await self._conn.execute(
"SELECT * FROM llm_models WHERE id = ?", (model_id,)
)
row = await cursor.fetchone()
return self._row_to_llm_model(row) if row else None
async def get_default_llm_model(self) -> Optional[LLMModel]:
"""Retrieve the default LLM model."""
cursor = await self._conn.execute(
"SELECT * FROM llm_models WHERE is_default = 1 LIMIT 1"
)
row = await cursor.fetchone()
return self._row_to_llm_model(row) if row else None
async def update_token_usage(
self,
model_id: str,
input_tokens: int,
output_tokens: int,
) -> None:
"""
Atomically increment token usage counters for a model.
Using SQL addition rather than read-modify-write prevents race
conditions when multiple agents use the same model concurrently.
Two concurrent tasks both reading 1000, adding 50, and writing
1050 would lose 50 tokens. The SQL approach makes the increment
atomic at the database level.
"""
await self._conn.execute(
"""
UPDATE llm_models
SET input_tokens_used = input_tokens_used + ?,
output_tokens_used = output_tokens_used + ?
WHERE id = ?
""",
(input_tokens, output_tokens, model_id),
)
await self._conn.commit()
async def delete_llm_model(self, model_id: str) -> None:
"""Delete an LLM model by ID."""
await self._conn.execute(
"DELETE FROM llm_models WHERE id = ?", (model_id,)
)
await self._conn.commit()
async def set_default_model(self, model_id: str) -> None:
"""
Set a model as the default, clearing the flag from all others.
Both updates are performed before committing so that there is
never a moment when no model is marked as default.
"""
await self._conn.execute(
"UPDATE llm_models SET is_default = 0"
)
await self._conn.execute(
"UPDATE llm_models SET is_default = 1 WHERE id = ?",
(model_id,),
)
await self._conn.commit()
def _row_to_llm_model(self, row: tuple) -> LLMModel:
"""Convert a database row tuple to an LLMModel instance."""
return LLMModel(
id=row[0],
display_name=row[1],
base_url=row[2],
model_name=row[3],
api_key=row[4],
is_default=bool(row[5]),
input_tokens_used=row[6],
output_tokens_used=row[7],
created_at=row[8],
)
# ------------------------------------------------------------------
# Session CRUD
# ------------------------------------------------------------------
async def save_session(self, session: Session) -> None:
"""Insert or replace a session record."""
await self._conn.execute(
"""
INSERT OR REPLACE INTO sessions
(id, title, model_id, is_agent_mode, messages,
is_active, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
session.id,
session.title,
session.model_id,
1 if session.is_agent_mode else 0,
json.dumps([m.model_dump() for m in session.messages]),
1 if session.is_active else 0,
session.created_at or _now_iso(),
_now_iso(),
),
)
await self._conn.commit()
async def get_all_sessions(self) -> list[Session]:
"""Retrieve all saved (inactive) sessions, newest first."""
cursor = await self._conn.execute(
"SELECT * FROM sessions WHERE is_active = 0 "
"ORDER BY updated_at DESC"
)
rows = await cursor.fetchall()
return [self._row_to_session(r) for r in rows]
async def get_session(self, session_id: str) -> Optional[Session]:
"""Retrieve a single session by ID."""
cursor = await self._conn.execute(
"SELECT * FROM sessions WHERE id = ?", (session_id,)
)
row = await cursor.fetchone()
return self._row_to_session(row) if row else None
async def delete_all_sessions(self) -> None:
"""Delete all sessions. Called when the user clears everything."""
await self._conn.execute("DELETE FROM sessions")
await self._conn.commit()
def _row_to_session(self, row: tuple) -> Session:
"""
Convert a database row to a Session instance.
The messages column is a JSON array of message dicts. Pydantic v2
automatically coerces nested dicts into the correct model types
(Message, ToolCall) during construction.
"""
raw_messages = json.loads(row[4])
messages = [Message(**m) for m in raw_messages]
return Session(
id=row[0],
title=row[1],
model_id=row[2],
is_agent_mode=bool(row[3]),
messages=messages,
is_active=bool(row[5]),
created_at=row[6],
updated_at=row[7],
)
# ------------------------------------------------------------------
# Agent Config CRUD
# ------------------------------------------------------------------
async def save_agent_config(self, agent: AgentConfig) -> None:
"""Insert or replace an agent configuration."""
await self._conn.execute(
"""
INSERT OR REPLACE INTO agent_configs
(id, name, description, system_prompt,
mcp_server_ids, max_iterations, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
agent.id,
agent.name,
agent.description,
agent.system_prompt,
json.dumps(agent.mcp_server_ids),
agent.max_iterations,
agent.created_at or _now_iso(),
),
)
await self._conn.commit()
async def get_all_agent_configs(self) -> list[AgentConfig]:
"""Retrieve all agent configurations, sorted by name."""
cursor = await self._conn.execute(
"SELECT * FROM agent_configs ORDER BY name ASC"
)
rows = await cursor.fetchall()
return [self._row_to_agent(r) for r in rows]
def _row_to_agent(self, row: tuple) -> AgentConfig:
"""Convert a database row to an AgentConfig instance."""
return AgentConfig(
id=row[0],
name=row[1],
description=row[2],
system_prompt=row[3],
mcp_server_ids=json.loads(row[4]),
max_iterations=row[5],
created_at=row[6],
)
# ------------------------------------------------------------------
# MCP Server CRUD
# ------------------------------------------------------------------
async def save_mcp_server(self, server: MCPServer) -> None:
"""Insert or replace an MCP server configuration."""
await self._conn.execute(
"""
INSERT OR REPLACE INTO mcp_servers
(id, name, transport, command, args, url, api_key,
env, is_active, cached_tools, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
server.id,
server.name,
server.transport,
server.command,
json.dumps(server.args),
server.url,
server.api_key,
json.dumps(server.env),
1 if server.is_active else 0,
json.dumps(server.cached_tools),
server.created_at or _now_iso(),
),
)
await self._conn.commit()
async def get_all_mcp_servers(self) -> list[MCPServer]:
"""Retrieve all configured MCP servers, sorted by name."""
cursor = await self._conn.execute(
"SELECT * FROM mcp_servers ORDER BY name ASC"
)
rows = await cursor.fetchall()
return [self._row_to_mcp_server(r) for r in rows]
async def get_mcp_server(
self, server_id: str
) -> Optional[MCPServer]:
"""Retrieve a single MCP server by ID."""
cursor = await self._conn.execute(
"SELECT * FROM mcp_servers WHERE id = ?", (server_id,)
)
row = await cursor.fetchone()
return self._row_to_mcp_server(row) if row else None
async def update_mcp_server_tools(
self, server_id: str, tools: list[dict]
) -> None:
"""
Persist the cached tool schemas for an MCP server.
This is called after the first successful tool discovery so
that subsequent agent runs can use the cached schemas without
re-connecting to the server.
"""
await self._conn.execute(
"UPDATE mcp_servers SET cached_tools = ? WHERE id = ?",
(json.dumps(tools), server_id),
)
await self._conn.commit()
async def delete_mcp_server(self, server_id: str) -> None:
"""Delete an MCP server configuration."""
await self._conn.execute(
"DELETE FROM mcp_servers WHERE id = ?", (server_id,)
)
await self._conn.commit()
def _row_to_mcp_server(self, row: tuple) -> MCPServer:
"""Convert a database row to an MCPServer instance."""
return MCPServer(
id=row[0],
name=row[1],
transport=row[2],
command=row[3],
args=json.loads(row[4]),
url=row[5],
api_key=row[6],
env=json.loads(row[7]),
is_active=bool(row[8]),
cached_tools=json.loads(row[9]),
created_at=row[10],
)
# Global database instance, initialised at startup via db.connect().
db = Database(config.database_path)
CHAPTER 5: THE LLM SERVICE - TALKING TO LOCAL AND REMOTE MODELS
The LLM service is the heart of the application. It abstracts away all differences between local and remote language models behind a single, clean interface. Whether you are talking to Ollama on your laptop or to GPT-4o on OpenAI's servers, the calling code is identical.
The key insight is that the OpenAI Chat Completions API has become the de facto standard for LLM communication. Ollama exposes it at /v1/chat/completions, LM Studio exposes it at /v1/chat/completions, vLLM exposes it at /v1/chat/completions, and of course OpenAI itself uses it. By targeting this single interface, the application works with all of them.
The service supports two modes: streaming and non-streaming. Streaming is used for simple chat sessions, delivering tokens to the user as they are produced. Non-streaming is used inside the ReACT agent loop, where the full response (including any tool call decisions) must be parsed before the next step can begin.
The stream_chat method uses httpx's async streaming context manager, which keeps the HTTP connection open and delivers response body bytes as they arrive. The SSE (Server-Sent Events) parsing loop processes each line: lines that start with "data: " contain JSON chunks, and the stream ends when a "[DONE]" sentinel is received. This is exactly how the OpenAI API works, and since all compatible servers follow the same convention, this single implementation works everywhere.
# backend/services/llm_service.py
"""
LLM communication service with streaming support.
This service handles all communication with language model endpoints.
It uses the OpenAI Chat Completions API format, which is supported by
Ollama, LM Studio, vLLM, OpenAI, Mistral, Groq, and most modern
LLM serving frameworks.
Streaming is implemented using Server-Sent Events (SSE) parsing,
which is the standard mechanism used by the OpenAI API and its
compatible implementations.
"""
import json
import logging
from collections.abc import AsyncGenerator
from typing import Optional
import httpx
from backend.models.llm_model import LLMModel
from backend.models.session import Message
logger = logging.getLogger(__name__)
# Generous timeouts for LLM calls. Local models on slower hardware
# may need the full read timeout for large outputs. The connect
# timeout is shorter because a connection failure should be detected
# quickly so the user gets a useful error message.
LLM_TIMEOUT = httpx.Timeout(
connect=10.0,
read=300.0,
write=30.0,
pool=5.0,
)
class LLMResponse:
"""
Encapsulates the result of a complete (non-streaming) LLM call.
Carries both the generated text and the token usage statistics
that are used to update the model's usage counters in the database.
The tool_calls list is populated when the LLM decides to call one
or more tools as part of the ReACT agent loop.
"""
def __init__(
self,
content: str,
input_tokens: int = 0,
output_tokens: int = 0,
tool_calls: Optional[list[dict]] = None,
) -> None:
self.content = content
self.input_tokens = input_tokens
self.output_tokens = output_tokens
self.tool_calls = tool_calls or []
class LLMService:
"""
Handles all LLM API communication.
This class is stateless; it receives the model configuration and
message history on each call. This makes it safe to call from
multiple concurrent coroutines without any locking.
"""
def _build_headers(self, model: LLMModel) -> dict[str, str]:
"""
Build HTTP headers for an LLM API request.
The Authorization header is only included when an API key is
configured, which allows unauthenticated local servers (like
default Ollama) to work without modification.
"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
}
if model.api_key:
headers["Authorization"] = f"Bearer {model.api_key}"
return headers
def _build_endpoint(self, model: LLMModel) -> str:
"""
Construct the full chat completions endpoint URL.
Ollama's OpenAI-compatible endpoint is at /v1/chat/completions.
OpenAI and most other services also use /v1/chat/completions.
The base_url should NOT include a trailing slash.
"""
base = model.base_url.rstrip("/")
return f"{base}/v1/chat/completions"
def _messages_to_api_format(
self, messages: list[Message]
) -> list[dict]:
"""
Convert internal Message objects to the OpenAI API wire format.
Tool call messages require special handling: the assistant
message that requested the tool call must include the
tool_calls array, and the subsequent tool result message
must reference the tool_call_id so the LLM can match results
to their originating requests.
"""
api_messages = []
for msg in messages:
if msg.role in ("user", "system"):
api_messages.append({
"role": msg.role,
"content": msg.content,
})
elif msg.role == "assistant":
entry: dict = {
"role": "assistant",
"content": msg.content,
}
if msg.tool_calls:
# Format tool calls in the OpenAI function-calling
# format that all compatible servers understand.
entry["tool_calls"] = [
{
"id": tc.tool_call_id,
"type": "function",
"function": {
"name": tc.tool_name,
"arguments": json.dumps(tc.arguments),
},
}
for tc in msg.tool_calls
]
api_messages.append(entry)
elif msg.role == "tool":
# Tool result messages reference the original call ID.
tool_call_id = (
msg.tool_calls[0].tool_call_id
if msg.tool_calls
else "unknown"
)
api_messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"content": msg.content,
})
return api_messages
async def stream_chat(
self,
model: LLMModel,
messages: list[Message],
tools: Optional[list[dict]] = None,
) -> AsyncGenerator[str, None]:
"""
Stream a chat completion response token by token.
This async generator yields text chunks as they arrive from the
LLM. The caller forwards these chunks to the WebSocket connection
for real-time display in the frontend.
The function handles the SSE (Server-Sent Events) format used
by the OpenAI API: each line starts with 'data: ' followed by
a JSON object, and the stream ends with 'data: [DONE]'.
"""
endpoint = self._build_endpoint(model)
headers = self._build_headers(model)
payload: dict = {
"model": model.model_name,
"messages": self._messages_to_api_format(messages),
"stream": True,
"temperature": 0.7,
}
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
async with httpx.AsyncClient(timeout=LLM_TIMEOUT) as client:
async with client.stream(
"POST",
endpoint,
headers=headers,
json=payload,
) as response:
response.raise_for_status()
async for line in response.aiter_lines():
line = line.strip()
if not line or not line.startswith("data: "):
continue
data_str = line[len("data: "):]
if data_str.strip() == "[DONE]":
break
try:
chunk = json.loads(data_str)
choices = chunk.get("choices", [])
if not choices:
continue
delta = choices[0].get("delta", {})
content = delta.get("content")
if content:
yield content
except json.JSONDecodeError:
# Malformed chunk from the server; skip and continue.
continue
async def complete_chat(
self,
model: LLMModel,
messages: list[Message],
tools: Optional[list[dict]] = None,
) -> LLMResponse:
"""
Perform a non-streaming chat completion and return the full
response including token usage statistics.
This method is used by the ReACT agent loop where we need the
complete response (including tool call decisions) before
proceeding to the next step. Streaming is not useful in the
agent loop because we need to parse the full response before
acting on it.
"""
endpoint = self._build_endpoint(model)
headers = self._build_headers(model)
payload: dict = {
"model": model.model_name,
"messages": self._messages_to_api_format(messages),
"stream": False,
"temperature": 0.7,
}
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
async with httpx.AsyncClient(timeout=LLM_TIMEOUT) as client:
response = await client.post(
endpoint, headers=headers, json=payload
)
response.raise_for_status()
data = response.json()
choice = data.get("choices", [{}])[0]
message = choice.get("message", {})
content = message.get("content") or ""
usage = data.get("usage", {})
# Parse tool calls from the response if present.
raw_tool_calls = message.get("tool_calls") or []
tool_calls = []
for tc in raw_tool_calls:
func = tc.get("function", {})
try:
args = json.loads(func.get("arguments", "{}"))
except json.JSONDecodeError:
args = {}
tool_calls.append({
"id": tc.get("id", ""),
"name": func.get("name", ""),
"arguments": args,
})
return LLMResponse(
content=content,
input_tokens=usage.get("prompt_tokens", 0),
output_tokens=usage.get("completion_tokens", 0),
tool_calls=tool_calls,
)
async def list_local_models(self, base_url: str) -> list[str]:
"""
Query an Ollama server for its list of available models.
This is a convenience method for the model configuration UI
that lets users discover what models are available locally
without having to type model names manually.
"""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.get(
f"{base_url.rstrip('/')}/api/tags"
)
response.raise_for_status()
data = response.json()
return [m["name"] for m in data.get("models", [])]
except Exception as exc:
logger.warning(
"Could not list local models from %s: %s", base_url, exc
)
return []
# Global service instance. Stateless, so safe to share across coroutines.
llm_service = LLMService()
CHAPTER 6: THE MCP SERVICE - CONNECTING TO TOOL SERVERS
The Model Context Protocol (MCP) is an open standard that defines how AI applications connect to external tools and data sources. The application uses FastMCP 3, the official high-level Python client library for MCP. FastMCP 3 abstracts away the entire transport layer, protocol handshake, and connection lifecycle, leaving the application code to focus purely on tool discovery and invocation.
FastMCP 3 provides a single unified Client class that works identically regardless of whether the underlying server communicates via stdio (a local subprocess) or HTTP (a remote service). The library handles the JSON-RPC 2.0 message framing, the MCP initialisation handshake, and connection health automatically. From the application's perspective, connecting to a local filesystem MCP server and connecting to a remote cloud-hosted MCP server look exactly the same.
An MCP server is a process or service that exposes a set of tools. Each tool has a name, a description, and a JSON Schema that defines its input parameters. When an agent wants to use a tool, it calls that tool by name with structured arguments. The MCP server executes the tool and returns a structured result. FastMCP 3 handles the entire wire protocol for this exchange.
The _mcp_tool_to_openai_format function is the bridge between the MCP world and the LLM world. FastMCP 3's list_tools() method returns Tool objects with name, description, and inputSchema attributes. The OpenAI function-calling format wraps this information in a specific structure with a "type": "function" envelope. This conversion is what allows the LLM to understand what tools are available and how to call them.
Tool schemas are cached on the MCPServer record after the first successful discovery. This avoids the overhead of re-connecting to the server and re-fetching the schema on every agent iteration. The cache is persisted to the database so it survives application restarts.
FastMCP 3's Client is used with a persistent connection pattern for efficiency. The client is entered into its async context once and kept alive for the duration of the application, which avoids the overhead of re-establishing the MCP handshake on every tool call. The shutdown method exits all active client contexts cleanly when the application stops.
# backend/services/mcp_service.py
"""
MCP (Model Context Protocol) service using FastMCP 3.
This service manages connections to MCP servers and provides a
unified interface for tool discovery and execution. It uses the
FastMCP 3 Client class, which handles both stdio-based local servers
and HTTP-based remote servers transparently.
FastMCP 3 manages the full MCP protocol lifecycle: transport
negotiation, the initialisation handshake, JSON-RPC message framing,
and connection health. The application code only needs to call
list_tools() and call_tool() on the Client instance.
Tool schemas discovered from MCP servers are cached on the server
record to avoid re-fetching them on every agent iteration.
"""
import logging
import os
from typing import Any, Optional
from fastmcp import Client
from fastmcp.client.transports import HttpTransport, StdioTransport
from backend.models.mcp_server import MCPServer
logger = logging.getLogger(__name__)
def _mcp_tool_to_openai_format(tool: Any) -> dict:
"""
Convert a FastMCP 3 Tool object to the OpenAI function-calling format.
The LLM receives tools in this format and uses it to decide which
tool to call and with what arguments.
FastMCP 3 Tool object attributes:
tool.name - str: the tool's identifier
tool.description - str: human-readable description
tool.inputSchema - dict: JSON Schema for the tool's parameters
Resulting OpenAI function format:
{
"type": "function",
"function": {
"name": "read_file",
"description": "Read a file from the filesystem",
"parameters": {
"type": "object",
"properties": { "path": { ... } },
"required": ["path"]
}
}
}
"""
# Support both FastMCP Tool objects (attribute access) and plain
# dicts (for cached tools loaded from the database).
if isinstance(tool, dict):
name = tool.get("name", "")
description = tool.get("description", "")
schema = tool.get(
"inputSchema",
{"type": "object", "properties": {}},
)
else:
name = tool.name
description = getattr(tool, "description", "") or ""
schema = (
getattr(tool, "inputSchema", None)
or getattr(tool, "parameters", None)
or {"type": "object", "properties": {}}
)
return {
"type": "function",
"function": {
"name": name,
"description": description,
"parameters": schema,
},
}
class MCPToolResult:
"""
The result of calling an MCP tool.
Carries the text output and an error flag for clean error handling
in the agent loop. When is_error is True, the content describes
what went wrong and the agent can decide how to recover.
"""
def __init__(self, content: str, is_error: bool = False) -> None:
self.content = content
self.is_error = is_error
class MCPService:
"""
High-level service for MCP tool server management using FastMCP 3.
This service maintains a registry of active FastMCP 3 Client
instances and provides methods for tool discovery and execution
that the agent service uses. Each client corresponds to one
configured MCP server and maintains a persistent connection for
the lifetime of the application.
FastMCP 3 handles all transport details internally: for stdio
servers it manages the subprocess lifecycle, and for HTTP servers
it manages the HTTP session. The application code is identical
for both transport types.
"""
def __init__(self) -> None:
# Maps server_id -> active FastMCP 3 Client instance.
# Clients are entered into their async context on first use
# and remain connected until shutdown() is called.
self._clients: dict[str, Client] = {}
async def _get_or_start_client(self, server: MCPServer) -> Client:
"""
Return an existing connected Client for the server, or create
and connect a new one.
For stdio servers, FastMCP 3 launches the subprocess and
completes the MCP initialisation handshake automatically.
For HTTP servers, FastMCP 3 establishes the HTTP session.
The client is stored after connection so that subsequent calls
reuse the same connection without repeating the handshake.
"""
if server.id in self._clients:
return self._clients[server.id]
if server.transport == "stdio":
if not server.command:
raise ValueError(
f"MCP server '{server.name}' has no command "
"configured for stdio transport."
)
# Merge the server's custom environment variables with the
# current process environment so the subprocess inherits
# PATH and other essential variables.
env = {**os.environ, **server.env}
transport = StdioTransport(
command=server.command,
args=server.args,
env=env,
)
else:
if not server.url:
raise ValueError(
f"MCP server '{server.name}' has no URL "
"configured for HTTP transport."
)
transport = HttpTransport(url=server.url)
client = Client(transport)
# Enter the async context manager to establish the connection.
# We store the connected client and exit the context on shutdown.
await client.__aenter__()
self._clients[server.id] = client
logger.info(
"FastMCP 3 client connected: %s (%s)",
server.name, server.transport,
)
return client
async def get_tools_for_servers(
self,
server_ids: list[str],
servers: list[MCPServer],
) -> list[dict]:
"""
Collect and return all tool schemas from the specified servers.
Uses cached schemas when available to avoid repeated round-trips.
When a server has no cached tools, connects to it via FastMCP 3,
fetches the tool list, converts to OpenAI format, and persists
the cache to the database.
"""
# Import here to avoid a circular import at module load time.
from backend.database import db
all_tools: list[dict] = []
server_map = {s.id: s for s in servers}
for server_id in server_ids:
server = server_map.get(server_id)
if not server or not server.is_active:
continue
if server.cached_tools:
# Use the in-memory cached schemas (already in OpenAI format).
all_tools.extend(server.cached_tools)
else:
try:
client = await self._get_or_start_client(server)
# FastMCP 3 returns list[Tool] with .name,
# .description, and .inputSchema attributes.
raw_tools = await client.list_tools()
tools = [
_mcp_tool_to_openai_format(t) for t in raw_tools
]
# Cache the schemas on the server object and in the DB.
server.cached_tools = tools
await db.update_mcp_server_tools(server_id, tools)
all_tools.extend(tools)
logger.info(
"Discovered %d tools from '%s'",
len(tools), server.name,
)
except Exception as exc:
logger.error(
"Failed to fetch tools from '%s': %s",
server.name, exc,
)
return all_tools
async def call_tool(
self,
server_id: str,
tool_name: str,
arguments: dict[str, Any],
servers: list[MCPServer],
) -> MCPToolResult:
"""
Execute a tool call on the appropriate MCP server via FastMCP 3.
FastMCP 3's call_tool() returns a list of Content objects.
Text content items have a .text attribute. Errors are surfaced
as Python exceptions, which are caught and returned as an
MCPToolResult with is_error=True so the agent can handle them.
"""
server_map = {s.id: s for s in servers}
server = server_map.get(server_id)
if not server:
return MCPToolResult(
content=f"MCP server '{server_id}' not found.",
is_error=True,
)
try:
client = await self._get_or_start_client(server)
# FastMCP 3 call_tool returns list[Content].
result_contents = await client.call_tool(
tool_name, arguments
)
text_parts: list[str] = []
is_error = False
for content_item in result_contents:
text = getattr(content_item, "text", None)
if text:
text_parts.append(text)
# Some MCP servers signal errors via the isError flag
# on individual content items.
if getattr(content_item, "isError", False):
is_error = True
return MCPToolResult(
content="\n".join(text_parts) or "(no output)",
is_error=is_error,
)
except Exception as exc:
logger.error(
"Tool call failed: %s/%s: %s",
server_id, tool_name, exc,
)
return MCPToolResult(
content=f"Error calling tool '{tool_name}': {exc}",
is_error=True,
)
async def find_server_for_tool(
self,
tool_name: str,
server_ids: list[str],
servers: list[MCPServer],
) -> Optional[str]:
"""
Find which server provides a given tool by name.
Searches the cached tool schemas of each server in the provided
list. Returns the server_id of the first server that has the
tool, or None if no server provides it.
"""
server_map = {s.id: s for s in servers}
for server_id in server_ids:
server = server_map.get(server_id)
if not server:
continue
for tool in server.cached_tools:
# Cached tools are in OpenAI format:
# {"type": "function", "function": {"name": ..., ...}}
func = tool.get("function", {})
if func.get("name") == tool_name:
return server_id
return None
async def shutdown(self) -> None:
"""
Disconnect all active FastMCP 3 clients cleanly.
Exits the async context manager for each client, which causes
FastMCP 3 to terminate stdio subprocesses and close HTTP
sessions gracefully.
"""
for server_id, client in list(self._clients.items()):
try:
await client.__aexit__(None, None, None)
logger.info(
"FastMCP 3 client disconnected: %s", server_id
)
except Exception as exc:
logger.warning(
"Error disconnecting client %s: %s", server_id, exc
)
self._clients.clear()
logger.info("All MCP clients shut down.")
# Global MCP service instance.
mcp_service = MCPService()
CHAPTER 7: THE REACT AGENT SERVICE
The ReACT pattern (Reasoning, Acting, Observing) is the algorithmic heart of the agent system. It was introduced in the paper "ReAct: Synergizing Reasoning and Acting in Language Models" and has become the dominant approach for building LLM-based agents. The pattern works as follows.
In the Reasoning phase, the LLM receives the conversation history and the list of available tools. It produces a response that either directly answers the user's question (in which case the loop ends) or decides to call a tool (in which case the loop continues). The LLM's reasoning is implicit in its response: it considers the goal, what it knows, and what tools might help.
In the Acting phase, if the LLM decided to call a tool, the application extracts the tool name and arguments from the LLM's response and sends the call to the appropriate MCP server.
In the Observing phase, the tool result is added to the conversation history as a "tool" message. The LLM can now see what the tool returned and use that information in its next reasoning step.
This loop continues until the LLM produces a response without any tool calls, which signals that it has reached a conclusion and is ready to present the final answer to the user.
The agent loop is designed around a critical principle: every step must be observable. The frontend needs to show the user what the agent is doing at each moment, not just the final answer. This is why the run method is an async generator that yields AgentEvent objects rather than simply returning a string. The caller can iterate over these events and forward them to the WebSocket connection in real time.
The iteration limit is an important safety mechanism. Without it, a confused agent could enter an infinite loop of tool calls, consuming tokens and time indefinitely. The limit is configurable per agent configuration, with a sensible default of 15 iterations. Most real-world tasks complete in 3 to 7 iterations.
# backend/services/agent_service.py
"""
ReACT agent service for autonomous goal-pursuit.
This service implements the ReACT (Reasoning, Acting, Observing)
pattern for LLM-based agents. It manages the agent loop, coordinates
tool calls through the MCP service, and streams progress updates
to the caller via an async generator.
The agent loop is designed to be fully observable: every step
(reasoning, tool call, tool result, final answer) is yielded as a
structured AgentEvent so the frontend can display real-time progress.
"""
import logging
from collections.abc import AsyncGenerator
from typing import Any
from backend.models.agent import AgentConfig
from backend.models.llm_model import LLMModel
from backend.models.mcp_server import MCPServer
from backend.models.session import Message, Session, ToolCall
from backend.services.llm_service import LLMResponse, llm_service
from backend.services.mcp_service import mcp_service
logger = logging.getLogger(__name__)
class AgentEvent:
"""
A structured event emitted during agent execution.
The type field identifies the event kind:
'iteration' - Iteration counter update (current / max).
'thinking' - The agent produced reasoning text.
'tool_call' - The agent is calling a tool.
'tool_result' - A tool returned a result.
'answer' - The agent has a final answer for the user.
'token_update'- Token usage statistics for the current LLM call.
'error' - An error occurred during the agent loop.
"""
def __init__(self, event_type: str, data: dict[str, Any]) -> None:
self.type = event_type
self.data = data
def to_dict(self) -> dict:
"""Serialise the event to a plain dict for JSON transmission."""
return {"event": self.type, "data": self.data}
class AgentService:
"""
Executes the ReACT agent loop for a given session and goal.
The run() method is an async generator that yields AgentEvent
objects as the agent progresses through its reasoning and acting
steps. This allows the caller to stream real-time updates to
the frontend without waiting for the entire agent run to complete.
"""
async def run(
self,
goal: str,
session: Session,
model: LLMModel,
agent_config: AgentConfig,
mcp_servers: list[MCPServer],
) -> AsyncGenerator[AgentEvent, None]:
"""
Execute the ReACT agent loop.
This generator drives the full agent lifecycle:
1. Prepare the initial message list with system prompt and goal.
2. Fetch available tools from configured MCP servers.
3. Enter the ReACT loop:
a. Call the LLM with current messages and tools (Reasoning).
b. If the LLM calls tools, execute them (Acting + Observing).
c. If the LLM gives a final answer, yield it and stop.
4. Yield events at each step for real-time frontend updates.
"""
# Collect tools from all MCP servers the agent can access.
tools = await mcp_service.get_tools_for_servers(
agent_config.mcp_server_ids, mcp_servers
)
# Build the initial message list with the system prompt.
messages: list[Message] = [
Message(
role="system",
content=agent_config.system_prompt,
),
]
# Include existing session history for multi-turn agent sessions.
# Skip any prior system messages to avoid duplication.
for msg in session.messages:
if msg.role != "system":
messages.append(msg)
# Add the user's goal as the latest user message.
user_message = Message(role="user", content=goal)
messages.append(user_message)
# Also record it in the session so it is persisted.
session.messages.append(user_message)
max_iterations = agent_config.max_iterations
for iteration in range(1, max_iterations + 1):
yield AgentEvent("iteration", {
"current": iteration,
"max": max_iterations,
})
logger.info(
"Agent iteration %d/%d for session %s",
iteration, max_iterations, session.id,
)
# ---- REASONING PHASE ----
# Call the LLM with the current message history and tools.
# We use the non-streaming complete_chat here because we
# need the full response (including tool call decisions)
# before we can proceed to the acting phase.
try:
llm_response: LLMResponse = await llm_service.complete_chat(
model=model,
messages=messages,
tools=tools if tools else None,
)
except Exception as exc:
logger.error(
"LLM call failed in agent loop: %s", exc
)
yield AgentEvent("error", {
"message": f"LLM call failed: {exc}"
})
return
# Emit token usage so the orchestrator can update the DB.
yield AgentEvent("token_update", {
"model_id": model.id,
"input_tokens": llm_response.input_tokens,
"output_tokens": llm_response.output_tokens,
})
# If the LLM produced text content, show it as reasoning.
if llm_response.content:
yield AgentEvent("thinking", {
"content": llm_response.content,
})
# ---- CHECK FOR FINAL ANSWER ----
# No tool calls means the LLM has reached a conclusion.
if not llm_response.tool_calls:
answer_message = Message(
role="assistant",
content=llm_response.content,
)
messages.append(answer_message)
session.messages.append(answer_message)
yield AgentEvent("answer", {
"content": llm_response.content,
})
return
# ---- ACTING PHASE ----
# The LLM wants to call one or more tools. Record the
# assistant's tool-calling message in the conversation history.
tool_calls_for_message = [
ToolCall(
tool_call_id=tc["id"],
tool_name=tc["name"],
arguments=tc["arguments"],
)
for tc in llm_response.tool_calls
]
assistant_message = Message(
role="assistant",
content=llm_response.content,
tool_calls=tool_calls_for_message,
)
messages.append(assistant_message)
session.messages.append(assistant_message)
# Execute each tool call and collect results.
for tc_dict in llm_response.tool_calls:
tool_name = tc_dict["name"]
tool_args = tc_dict["arguments"]
tool_call_id = tc_dict["id"]
yield AgentEvent("tool_call", {
"tool_name": tool_name,
"arguments": tool_args,
})
# ---- OBSERVING PHASE ----
# Find which MCP server provides this tool and call it.
server_id = await mcp_service.find_server_for_tool(
tool_name,
agent_config.mcp_server_ids,
mcp_servers,
)
if server_id:
tool_result = await mcp_service.call_tool(
server_id=server_id,
tool_name=tool_name,
arguments=tool_args,
servers=mcp_servers,
)
result_content = tool_result.content
is_error = tool_result.is_error
else:
result_content = (
f"No MCP server found that provides "
f"the tool '{tool_name}'."
)
is_error = True
yield AgentEvent("tool_result", {
"tool_name": tool_name,
"result": result_content,
"is_error": is_error,
})
# Add the tool result to the conversation history.
# The tool_call_id links this result to the request
# that generated it, which is required by the protocol.
tool_result_message = Message(
role="tool",
content=result_content,
tool_calls=[
ToolCall(
tool_call_id=tool_call_id,
tool_name=tool_name,
arguments=tool_args,
result=result_content if not is_error else None,
error=result_content if is_error else None,
)
],
)
messages.append(tool_result_message)
session.messages.append(tool_result_message)
# If we exit the loop, the agent hit the iteration limit.
logger.warning(
"Agent hit max iterations (%d) for session %s",
max_iterations, session.id,
)
yield AgentEvent("error", {
"message": (
f"Agent reached the maximum of {max_iterations} "
"iterations without a final answer. The task may be "
"too complex or the agent may be stuck in a loop."
)
})
# Global agent service instance.
agent_service = AgentService()
CHAPTER 8: THE SESSION SERVICE
The session service manages the lifecycle of conversation sessions. It handles creating new sessions, adding messages to active sessions, saving completed sessions to the database, and retrieving session history. It also generates meaningful titles for sessions based on the first user message, which makes the sessions list easy to navigate.
The _active_sessions dictionary is an instance variable (not a class variable), which is the correct Python pattern. Class variables are shared across all instances of a class, which would cause subtle bugs if multiple Database or SessionService instances were ever created. Instance variables, initialised in init, belong exclusively to one instance.
# backend/services/session_service.py
"""
Session lifecycle management service.
This service handles all operations on conversation sessions:
creating new sessions, appending messages, saving completed sessions
to the database, and retrieving session history. It also generates
concise, meaningful titles for sessions based on the first user
message, which makes the sessions list easy to navigate.
"""
import logging
from datetime import datetime, timezone
from typing import Optional
from backend.database import db
from backend.models.session import Message, Session
logger = logging.getLogger(__name__)
def _now_iso() -> str:
"""Return the current UTC time as an ISO 8601 string."""
return datetime.now(timezone.utc).isoformat()
def _generate_title(first_user_message: str) -> str:
"""
Generate a concise session title from the first user message.
Collapses whitespace and truncates long messages with an ellipsis.
This gives the sessions list meaningful, scannable entries without
requiring the user to name their sessions manually.
"""
cleaned = " ".join(first_user_message.split())
if len(cleaned) <= 50:
return cleaned
return cleaned[:47] + "..."
class SessionService:
"""
Manages the full lifecycle of conversation sessions.
Sessions are created when a new tab is opened, updated as messages
are added, and saved when a tab is closed. The service also provides
methods for retrieving saved sessions for display in the sessions
list panel.
Active sessions are kept in memory for fast access. They are also
persisted to the database on every update so that they survive an
unexpected application restart.
"""
def __init__(self) -> None:
# In-memory store of active sessions, keyed by session_id.
# Active sessions live here until the tab is closed, at which
# point they are marked inactive and remain only in the database.
self._active_sessions: dict[str, Session] = {}
async def create_session(
self,
model_id: str,
is_agent_mode: bool = False,
tab_id: Optional[str] = None,
) -> Session:
"""
Create a new session and register it as active.
The session is immediately saved to the database so it survives
application restarts even if the tab has not been closed yet.
"""
session = Session(
model_id=model_id,
is_agent_mode=is_agent_mode,
tab_id=tab_id,
is_active=True,
created_at=_now_iso(),
updated_at=_now_iso(),
)
self._active_sessions[session.id] = session
await db.save_session(session)
logger.info(
"Created session %s (agent=%s)", session.id, is_agent_mode
)
return session
def get_active_session(
self, session_id: str
) -> Optional[Session]:
"""Retrieve an active (in-memory) session by ID."""
return self._active_sessions.get(session_id)
async def add_message(
self,
session_id: str,
message: Message,
) -> Optional[Session]:
"""
Append a message to an active session and persist the update.
Generates a session title from the first user message if the
session still has the default 'New Chat' title. Returns the
updated session, or None if the session is not found.
"""
session = self._active_sessions.get(session_id)
if not session:
logger.warning(
"add_message: session %s not found", session_id
)
return None
# Generate a title from the first user message.
if message.role == "user" and session.title == "New Chat":
session.title = _generate_title(message.content)
session.messages.append(message)
session.updated_at = _now_iso()
await db.save_session(session)
return session
async def close_session(
self, session_id: str
) -> Optional[Session]:
"""
Mark a session as inactive (closed tab) and persist it.
The session moves from the active in-memory store to the saved
sessions list in the database. It remains queryable via
get_session_detail for display in the sessions list modal.
"""
session = self._active_sessions.pop(session_id, None)
if not session:
return None
session.is_active = False
session.tab_id = None
session.updated_at = _now_iso()
await db.save_session(session)
logger.info("Closed and saved session %s", session_id)
return session
async def get_saved_sessions(self) -> list[Session]:
"""
Retrieve all saved (closed) sessions from the database.
Returns them ordered by most recently updated first.
"""
return await db.get_all_sessions()
async def get_session_detail(
self, session_id: str
) -> Optional[Session]:
"""
Retrieve the full detail of a session.
Checks active in-memory sessions first (for currently open tabs),
then falls back to the database (for saved sessions).
"""
active = self._active_sessions.get(session_id)
if active:
return active
return await db.get_session(session_id)
async def delete_all(self) -> None:
"""
Delete all sessions, both active and saved.
Called when the user chooses to clear everything.
"""
self._active_sessions.clear()
await db.delete_all_sessions()
logger.info("All sessions deleted.")
async def update_token_usage(
self,
model_id: str,
input_tokens: int,
output_tokens: int,
) -> None:
"""
Update token usage counters for a model.
Delegates to the database layer for an atomic SQL update.
"""
await db.update_token_usage(model_id, input_tokens, output_tokens)
# Global session service instance.
session_service = SessionService()
CHAPTER 9: THE ORCHESTRATOR - MANAGING CONCURRENT AGENTS
The orchestrator is the component that makes concurrent agent execution possible. It manages a pool of running tasks, one per active tab, and provides a clean interface for starting, monitoring, and stopping them. Each task is an asyncio Task that runs the agent loop or a simple chat stream for its respective tab.
The orchestrator is what distinguishes this application from a simple sequential chatbot. When the user opens three tabs and starts a prompt in each, the orchestrator creates three concurrent asyncio tasks. These tasks run on the same event loop, interleaved by asyncio's cooperative multitasking. When one task is waiting for an LLM response (an I/O-bound operation), the event loop switches to another task and makes progress there. This is why asyncio is perfect for this use case: LLM calls are network I/O, and asyncio excels at managing many concurrent network operations efficiently.
The cancellation mechanism is equally important. When a user closes a tab or sends a new message while one is still generating, the orchestrator cancels the running task cleanly. asyncio.CancelledError is raised inside the task at the next await point, the task's finally block runs to clean up, and the task is removed from the registry.
The token estimation in the streaming path uses all messages that were sent to the LLM as the basis for input token estimation. This is the correct set: all messages in the session at the time of the LLM call, including the user's latest message.
# backend/orchestrator.py
"""
Concurrent agent and chat session orchestrator.
This module manages the lifecycle of all active chat and agent tasks.
Each tab in the frontend corresponds to one asyncio Task in the
orchestrator. Tasks run concurrently on the event loop, interleaved
by cooperative multitasking at every await point.
The orchestrator uses a WebSocket callback pattern: when a task
produces output (a streamed token, an agent event, or an error),
it calls the registered callback for that tab's WebSocket connection.
This decouples task execution from WebSocket management.
"""
import asyncio
import logging
from collections.abc import Callable, Coroutine
from typing import Any, Optional
from backend.database import db
from backend.models.agent import AgentConfig
from backend.models.llm_model import LLMModel
from backend.models.mcp_server import MCPServer
from backend.models.session import Message, Session
from backend.services.agent_service import AgentEvent, agent_service
from backend.services.llm_service import llm_service
from backend.services.session_service import session_service
logger = logging.getLogger(__name__)
# Type alias for the WebSocket send callback.
# The callback receives a dict and sends it as JSON to the client.
SendCallback = Callable[[dict], Coroutine[Any, Any, None]]
class TabTask:
"""
Represents a single active tab's running task.
Each TabTask holds a reference to the asyncio Task running the
chat or agent loop, the session it belongs to, and the WebSocket
send callback for delivering output to the frontend.
"""
def __init__(
self,
tab_id: str,
session_id: str,
task: asyncio.Task,
send: SendCallback,
) -> None:
self.tab_id = tab_id
self.session_id = session_id
self.task = task
self.send = send
class Orchestrator:
"""
Manages all concurrent chat and agent tasks.
The orchestrator is the central coordinator of the application.
It receives requests from the API layer to start new tasks,
cancel running tasks, and clean up completed tasks.
"""
def __init__(self) -> None:
# Maps tab_id -> TabTask for all currently running tabs.
self._tasks: dict[str, TabTask] = {}
async def start_chat_task(
self,
tab_id: str,
session_id: str,
user_message: str,
model: LLMModel,
send: SendCallback,
) -> None:
"""
Start a streaming chat task for a simple (non-agent) tab.
Creates an asyncio Task that streams the LLM response token by
token, calling the send callback for each chunk. The task runs
concurrently with all other active tasks.
"""
# Cancel any existing task for this tab (e.g., if the user
# sends a new message before the previous one finished).
await self._cancel_tab_task(tab_id)
session = session_service.get_active_session(session_id)
if not session:
await send({
"type": "error",
"tab_id": tab_id,
"message": "Session not found.",
})
return
# Add the user message to the session before starting the task.
user_msg = Message(role="user", content=user_message)
await session_service.add_message(session_id, user_msg)
task = asyncio.create_task(
self._run_chat_stream(
tab_id, session_id, session, model, send
),
name=f"chat-{tab_id}",
)
self._tasks[tab_id] = TabTask(
tab_id=tab_id,
session_id=session_id,
task=task,
send=send,
)
async def _run_chat_stream(
self,
tab_id: str,
session_id: str,
session: Session,
model: LLMModel,
send: SendCallback,
) -> None:
"""
Stream an LLM chat response to the frontend.
This coroutine runs as an asyncio Task. It streams tokens from
the LLM and sends each chunk to the frontend via the WebSocket
callback. When streaming completes, it saves the full response
as an assistant message in the session and updates token usage.
"""
full_response: list[str] = []
try:
await send({"type": "stream_start", "tab_id": tab_id})
# session.messages at this point includes the user message
# that was added before this task was created.
async for chunk in llm_service.stream_chat(
model=model,
messages=session.messages,
):
full_response.append(chunk)
await send({
"type": "stream_chunk",
"tab_id": tab_id,
"content": chunk,
})
# Streaming complete. Assemble and save the full response.
complete_text = "".join(full_response)
assistant_msg = Message(
role="assistant",
content=complete_text,
)
await session_service.add_message(session_id, assistant_msg)
# Estimate token usage for the streaming call.
# Input tokens: all messages sent to the LLM (the full
# session.messages list before the assistant response was added).
# We use a word-based approximation (1 word ~ 1.33 tokens).
estimated_input = sum(
len(m.content.split()) * 4 // 3
for m in session.messages
if m.role != "assistant" or m != assistant_msg
)
estimated_output = len(complete_text.split()) * 4 // 3
await db.update_token_usage(
model.id, estimated_input, estimated_output
)
await send({
"type": "stream_end",
"tab_id": tab_id,
"session_id": session_id,
"input_tokens": estimated_input,
"output_tokens": estimated_output,
})
except asyncio.CancelledError:
await send({
"type": "stream_cancelled",
"tab_id": tab_id,
})
raise
except Exception as exc:
logger.error(
"Chat stream error for tab %s: %s", tab_id, exc
)
await send({
"type": "error",
"tab_id": tab_id,
"message": str(exc),
})
finally:
# Always clean up the task reference when done, whether
# the task completed normally, was cancelled, or errored.
self._tasks.pop(tab_id, None)
async def start_agent_task(
self,
tab_id: str,
session_id: str,
goal: str,
model: LLMModel,
agent_config: AgentConfig,
mcp_servers: list[MCPServer],
send: SendCallback,
) -> None:
"""
Start a ReACT agent task for an agent-mode tab.
Creates an asyncio Task that runs the full agent loop, yielding
events at each step and forwarding them to the frontend via the
WebSocket callback.
"""
await self._cancel_tab_task(tab_id)
session = session_service.get_active_session(session_id)
if not session:
await send({
"type": "error",
"tab_id": tab_id,
"message": "Session not found.",
})
return
task = asyncio.create_task(
self._run_agent_loop(
tab_id, session_id, session, goal,
model, agent_config, mcp_servers, send,
),
name=f"agent-{tab_id}",
)
self._tasks[tab_id] = TabTask(
tab_id=tab_id,
session_id=session_id,
task=task,
send=send,
)
async def _run_agent_loop(
self,
tab_id: str,
session_id: str,
session: Session,
goal: str,
model: LLMModel,
agent_config: AgentConfig,
mcp_servers: list[MCPServer],
send: SendCallback,
) -> None:
"""
Run the ReACT agent loop and stream events to the frontend.
This coroutine drives the agent_service.run() generator,
forwarding each AgentEvent to the WebSocket as a structured
JSON message. Token usage updates are persisted to the database
as they arrive so the model list stays current.
"""
try:
await send({"type": "agent_start", "tab_id": tab_id})
async for event in agent_service.run(
goal=goal,
session=session,
model=model,
agent_config=agent_config,
mcp_servers=mcp_servers,
):
# Persist token usage updates immediately so the
# frontend can show accurate counts in real time.
if event.type == "token_update":
await db.update_token_usage(
model_id=event.data["model_id"],
input_tokens=event.data["input_tokens"],
output_tokens=event.data["output_tokens"],
)
# Forward all events to the frontend.
await send({
"type": "agent_event",
"tab_id": tab_id,
"event": event.to_dict(),
})
# Save the session after the agent completes so all the
# reasoning steps and tool calls are persisted.
await db.save_session(session)
await send({
"type": "agent_complete",
"tab_id": tab_id,
"session_id": session_id,
})
except asyncio.CancelledError:
await send({
"type": "agent_cancelled",
"tab_id": tab_id,
})
raise
except Exception as exc:
logger.error(
"Agent loop error for tab %s: %s", tab_id, exc
)
await send({
"type": "error",
"tab_id": tab_id,
"message": str(exc),
})
finally:
self._tasks.pop(tab_id, None)
async def cancel_tab(self, tab_id: str) -> bool:
"""
Cancel the running task for a specific tab.
Returns True if a task was found and cancelled.
"""
return await self._cancel_tab_task(tab_id)
async def _cancel_tab_task(self, tab_id: str) -> bool:
"""Cancel and clean up a tab task. Returns True if one existed."""
tab_task = self._tasks.pop(tab_id, None)
if not tab_task:
return False
if not tab_task.task.done():
tab_task.task.cancel()
try:
await tab_task.task
except asyncio.CancelledError:
pass
return True
def get_active_tab_ids(self) -> list[str]:
"""Return the IDs of all currently running tabs."""
return list(self._tasks.keys())
async def shutdown(self) -> None:
"""Cancel all running tasks during application shutdown."""
for tab_id in list(self._tasks.keys()):
await self._cancel_tab_task(tab_id)
logger.info("Orchestrator shut down.")
# Global orchestrator instance.
orchestrator = Orchestrator()
CHAPTER 10: THE FASTAPI BACKEND
The FastAPI backend wires all the services together and exposes them to the frontend. It uses WebSockets for real-time bidirectional communication (streaming LLM responses and agent events) and REST endpoints for configuration management (adding models, MCP servers, agent configurations, etc.).
The WebSocket endpoint is the most important part of the API. It receives all messages from the frontend, dispatches them to the appropriate handler, and sends responses back. The key design decision is to use a single WebSocket connection for all tabs, with the tab_id field multiplexing messages between them. This is more efficient than opening a separate WebSocket connection per tab and avoids connection management complexity in the frontend.
The send callback pattern is elegant: the orchestrator receives a callable that it can call at any time to send data to the frontend. This means the orchestrator tasks do not need to hold a reference to the WebSocket object, which could cause issues if the WebSocket disconnects while a task is running.
The lifespan context manager handles startup and shutdown cleanly. On startup, the database connection is opened and the schema is created. On shutdown, all running tasks are cancelled, all MCP server processes are stopped via FastMCP 3's client shutdown, and the database connection is closed. This ensures clean resource cleanup even when the application is stopped with Ctrl+C.
# backend/main.py
"""
FastAPI application entry point and API layer.
This module creates the FastAPI application, configures CORS,
registers all routes, and manages application lifecycle events.
It is the composition root of the application: all services and
the database are initialised here and wired together.
"""
import asyncio
import json
import logging
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Optional
import httpx
from fastapi import (
FastAPI,
HTTPException,
WebSocket,
WebSocketDisconnect,
status,
)
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from backend.config import config
from backend.database import db
from backend.models.agent import AgentConfig, DEFAULT_AGENT_SYSTEM_PROMPT
from backend.models.llm_model import LLMModel
from backend.models.mcp_server import MCPServer
from backend.orchestrator import orchestrator
from backend.services.mcp_service import mcp_service
from backend.services.session_service import session_service
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Application lifespan manager.
On startup: connect to the database and initialise the schema.
On shutdown: cancel all running tasks, disconnect FastMCP 3 clients,
and close the database connection.
"""
logger.info("Starting %s v%s", config.app_name, config.version)
await db.connect()
yield
logger.info("Shutting down %s...", config.app_name)
await orchestrator.shutdown()
await mcp_service.shutdown()
await db.disconnect()
app = FastAPI(
title=config.app_name,
version=config.version,
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=config.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Serve the frontend static files from the /static path.
app.mount(
"/static",
StaticFiles(directory="frontend"),
name="static",
)
# ------------------------------------------------------------------
# Request / Response Models
# ------------------------------------------------------------------
class AddLLMModelRequest(BaseModel):
"""Request body for adding a new LLM model."""
display_name: str
base_url: str
model_name: str
api_key: str = ""
is_default: bool = False
class AddMCPServerRequest(BaseModel):
"""Request body for adding a new MCP server."""
name: str
transport: str = "stdio"
command: Optional[str] = None
args: list[str] = []
url: Optional[str] = None
api_key: Optional[str] = None
env: dict[str, str] = {}
class AddAgentConfigRequest(BaseModel):
"""Request body for creating an agent configuration."""
name: str
description: str = ""
system_prompt: str = ""
mcp_server_ids: list[str] = []
max_iterations: int = 15
# ------------------------------------------------------------------
# REST Endpoints
# ------------------------------------------------------------------
@app.get("/")
async def serve_frontend() -> FileResponse:
"""Serve the main frontend HTML file."""
return FileResponse("frontend/index.html")
@app.get("/api/models")
async def get_models() -> list[dict]:
"""Return all configured LLM models with their token usage."""
models = await db.get_all_llm_models()
return [m.model_dump() for m in models]
@app.post("/api/models", status_code=status.HTTP_201_CREATED)
async def add_model(req: AddLLMModelRequest) -> dict:
"""Add a new LLM model configuration."""
now = datetime.now(timezone.utc).isoformat()
model = LLMModel(
display_name=req.display_name,
base_url=req.base_url,
model_name=req.model_name,
api_key=req.api_key,
is_default=req.is_default,
created_at=now,
)
if req.is_default:
# Clear the default flag from all other models first.
await db.set_default_model(model.id)
await db.save_llm_model(model)
return model.model_dump()
@app.delete("/api/models/{model_id}")
async def delete_model(model_id: str) -> dict:
"""Delete an LLM model configuration."""
await db.delete_llm_model(model_id)
return {"status": "deleted"}
@app.put("/api/models/{model_id}/set-default")
async def set_default_model(model_id: str) -> dict:
"""Set a model as the application default."""
await db.set_default_model(model_id)
return {"status": "updated"}
@app.get("/api/sessions")
async def get_sessions() -> list[dict]:
"""Return summary information for all saved (closed) sessions."""
sessions = await db.get_all_sessions()
return [
{
"id": s.id,
"title": s.title,
"model_id": s.model_id,
"is_agent_mode": s.is_agent_mode,
"message_count": len(s.messages),
"created_at": s.created_at,
"updated_at": s.updated_at,
}
for s in sessions
]
@app.get("/api/sessions/{session_id}")
async def get_session_detail(session_id: str) -> dict:
"""Return the full message history of a session."""
session = await session_service.get_session_detail(session_id)
if not session:
raise HTTPException(
status_code=404, detail="Session not found"
)
return session.model_dump()
@app.delete("/api/sessions")
async def delete_all_sessions() -> dict:
"""Delete all sessions (clear everything)."""
await session_service.delete_all()
return {"status": "all sessions deleted"}
@app.get("/api/mcp-servers")
async def get_mcp_servers() -> list[dict]:
"""Return all configured MCP servers."""
servers = await db.get_all_mcp_servers()
return [s.model_dump() for s in servers]
@app.post("/api/mcp-servers", status_code=status.HTTP_201_CREATED)
async def add_mcp_server(req: AddMCPServerRequest) -> dict:
"""Add a new MCP server configuration."""
now = datetime.now(timezone.utc).isoformat()
server = MCPServer(
name=req.name,
transport=req.transport,
command=req.command,
args=req.args,
url=req.url,
api_key=req.api_key,
env=req.env,
created_at=now,
)
await db.save_mcp_server(server)
return server.model_dump()
@app.delete("/api/mcp-servers/{server_id}")
async def delete_mcp_server(server_id: str) -> dict:
"""Delete an MCP server configuration."""
await db.delete_mcp_server(server_id)
return {"status": "deleted"}
@app.get("/api/agent-configs")
async def get_agent_configs() -> list[dict]:
"""Return all agent configurations."""
configs = await db.get_all_agent_configs()
return [c.model_dump() for c in configs]
@app.post("/api/agent-configs", status_code=status.HTTP_201_CREATED)
async def add_agent_config(req: AddAgentConfigRequest) -> dict:
"""Create a new agent configuration."""
now = datetime.now(timezone.utc).isoformat()
agent_cfg = AgentConfig(
name=req.name,
description=req.description,
# Use the module-level constant as the fallback so we never
# depend on Pydantic internals to retrieve a default value.
system_prompt=req.system_prompt or DEFAULT_AGENT_SYSTEM_PROMPT,
mcp_server_ids=req.mcp_server_ids,
max_iterations=req.max_iterations,
created_at=now,
)
await db.save_agent_config(agent_cfg)
return agent_cfg.model_dump()
@app.get("/api/health")
async def health_check() -> dict:
"""
Check connectivity to all configured LLM models.
Pings each model's /v1/models endpoint concurrently and returns
a status dict showing which models are reachable. Also reports
the IDs of all currently active tabs.
"""
models = await db.get_all_llm_models()
async def check_model(model: LLMModel) -> tuple[str, bool]:
"""Ping a model endpoint and return its reachability."""
try:
endpoint = f"{model.base_url.rstrip('/')}/v1/models"
headers: dict[str, str] = {}
if model.api_key:
headers["Authorization"] = f"Bearer {model.api_key}"
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(endpoint, headers=headers)
return model.id, resp.status_code < 400
except Exception:
return model.id, False
# Check all models concurrently for a fast response.
results_raw = await asyncio.gather(
*[check_model(m) for m in models],
return_exceptions=True,
)
model_status: dict[str, bool] = {}
for result in results_raw:
if isinstance(result, tuple):
model_id, is_healthy = result
model_status[model_id] = is_healthy
return {
"status": "ok",
"models": model_status,
"active_tabs": orchestrator.get_active_tab_ids(),
}
# ------------------------------------------------------------------
# WebSocket Endpoint
# ------------------------------------------------------------------
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
"""
Main WebSocket endpoint for real-time communication.
All tab interactions (sending messages, receiving streamed
responses, agent events) go through this single WebSocket
connection. The frontend multiplexes multiple tabs over one
connection using the tab_id field in each message.
Messages from client to server:
{ "action": "create_session", "tab_id": "...", ... }
{ "action": "send_message", "tab_id": "...", ... }
{ "action": "close_tab", "tab_id": "...", ... }
{ "action": "cancel_tab", "tab_id": "...", ... }
Messages from server to client:
{ "type": "session_created", "tab_id": "...", ... }
{ "type": "stream_chunk", "tab_id": "...", ... }
{ "type": "agent_event", "tab_id": "...", ... }
{ "type": "error", "tab_id": "...", ... }
"""
await websocket.accept()
logger.info("WebSocket connected: %s", websocket.client)
async def send(data: dict) -> None:
"""
Send a JSON message to the connected client.
This callback is passed to the orchestrator so tasks can send
output to the frontend without holding a direct reference to
the WebSocket object. Errors are silently swallowed because
the client may have disconnected.
"""
try:
await websocket.send_json(data)
except Exception:
pass
try:
while True:
raw = await websocket.receive_text()
try:
msg = json.loads(raw)
except json.JSONDecodeError:
await send({"type": "error", "message": "Invalid JSON"})
continue
action = msg.get("action", "")
tab_id = msg.get("tab_id", "")
if action == "create_session":
await _handle_create_session(msg, tab_id, send)
elif action == "send_message":
await _handle_send_message(msg, tab_id, send)
elif action == "close_tab":
await _handle_close_tab(msg, tab_id, send)
elif action == "cancel_tab":
await orchestrator.cancel_tab(tab_id)
await send({"type": "tab_cancelled", "tab_id": tab_id})
else:
await send({
"type": "error",
"message": f"Unknown action: '{action}'",
})
except WebSocketDisconnect:
logger.info("WebSocket disconnected: %s", websocket.client)
except Exception as exc:
logger.error("WebSocket error: %s", exc)
async def _handle_create_session(
msg: dict, tab_id: str, send
) -> None:
"""
Handle a 'create_session' WebSocket action.
Creates a new session and responds with the session details so
the frontend can associate the tab with its server-side session.
"""
model_id: Optional[str] = msg.get("model_id")
is_agent_mode: bool = msg.get("is_agent_mode", False)
# Use the default model if none was specified.
if not model_id:
default_model = await db.get_default_llm_model()
model_id = default_model.id if default_model else ""
session = await session_service.create_session(
model_id=model_id,
is_agent_mode=is_agent_mode,
tab_id=tab_id,
)
await send({
"type": "session_created",
"tab_id": tab_id,
"session_id": session.id,
"model_id": model_id,
"is_agent_mode": is_agent_mode,
})
async def _handle_send_message(
msg: dict, tab_id: str, send
) -> None:
"""
Handle a 'send_message' WebSocket action.
Routes to either the streaming chat path or the ReACT agent loop
depending on the session's current mode. The agent mode can be
toggled per-message by the frontend, and the session is updated
accordingly before routing.
"""
session_id: str = msg.get("session_id", "")
content: str = (msg.get("content") or "").strip()
model_id: Optional[str] = msg.get("model_id")
# The frontend sends the current state of the agent mode checkbox.
is_agent_mode: bool = msg.get("is_agent_mode", False)
if not content:
await send({
"type": "error",
"tab_id": tab_id,
"message": "Empty message.",
})
return
# Resolve the LLM model to use for this message.
if model_id:
model = await db.get_llm_model(model_id)
else:
model = await db.get_default_llm_model()
if not model:
await send({
"type": "error",
"tab_id": tab_id,
"message": "No LLM model configured. Please add one.",
})
return
session = session_service.get_active_session(session_id)
if not session:
await send({
"type": "error",
"tab_id": tab_id,
"message": f"Session '{session_id}' not found.",
})
return
# Sync the session's agent mode with the frontend's current state.
session.is_agent_mode = is_agent_mode
if is_agent_mode:
# Agent mode: resolve the agent config and run the ReACT loop.
agent_config_id: Optional[str] = msg.get("agent_config_id")
agent_cfg: Optional[AgentConfig] = None
if agent_config_id:
agent_cfgs = await db.get_all_agent_configs()
agent_cfg = next(
(a for a in agent_cfgs if a.id == agent_config_id),
None,
)
if not agent_cfg:
# Fall back to the first configured agent, or create a
# minimal default inline if none have been configured.
agent_cfgs = await db.get_all_agent_configs()
agent_cfg = agent_cfgs[0] if agent_cfgs else AgentConfig(
name="Default Agent",
mcp_server_ids=[],
)
mcp_servers = await db.get_all_mcp_servers()
await orchestrator.start_agent_task(
tab_id=tab_id,
session_id=session_id,
goal=content,
model=model,
agent_config=agent_cfg,
mcp_servers=mcp_servers,
send=send,
)
else:
# Simple chat mode: stream the response token by token.
await orchestrator.start_chat_task(
tab_id=tab_id,
session_id=session_id,
user_message=content,
model=model,
send=send,
)
async def _handle_close_tab(
msg: dict, tab_id: str, send
) -> None:
"""
Handle a 'close_tab' WebSocket action.
Cancels any running task for the tab, saves the session to the
database, and notifies the client that the tab has been closed.
"""
session_id: str = msg.get("session_id", "")
await orchestrator.cancel_tab(tab_id)
saved_session = await session_service.close_session(session_id)
await send({
"type": "tab_closed",
"tab_id": tab_id,
"session_id": session_id,
"title": saved_session.title if saved_session else "",
})
CHAPTER 11: THE FRONTEND
The frontend is a single-page web application written in vanilla JavaScript. It communicates with the backend over a single WebSocket connection and uses REST endpoints for configuration operations. The UI presents the main chat window with a tab bar at the top, a sessions list panel on the left, a model management panel on the right, and the active chat area in the centre.
The frontend architecture follows a clear separation of concerns. All UI state lives in a central state object. Rendering functions translate state into DOM updates. WebSocket messages from the server drive state changes, which trigger targeted DOM updates. This one-way data flow makes the application predictable and easy to debug.
The optimistic update pattern in sendMessage is worth highlighting. When the user clicks Send, the message is immediately added to the local state and rendered in the UI, before the server has even acknowledged it. This gives the user instant visual feedback and makes the application feel responsive even over slow connections.
The HTML structure is deliberately minimal. All dynamic content is managed by JavaScript. This makes the HTML easy to read and understand at a glance.
<!-- frontend/index.html -->
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Agentic Chatbot Orchestrator</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<!--
The application is divided into three main panels:
Left sidebar: saved sessions list
Centre: tab bar + active chat area + input
Right sidebar: model management + MCP server management
-->
<div id="app">
<!-- Left Sidebar: Saved Sessions -->
<aside id="sessions-panel">
<div class="panel-header">
<h2>Sessions</h2>
<button id="btn-clear-all" title="Delete all sessions">
Clear All
</button>
</div>
<div id="sessions-list">
<!-- Session items are injected here by JavaScript -->
</div>
</aside>
<!-- Centre: Main Chat Area -->
<main id="chat-area">
<!-- Tab Bar -->
<div id="tab-bar">
<div id="tabs-container">
<!-- Tab buttons are injected here by JavaScript -->
</div>
<button id="btn-new-tab" title="Open new chat tab">+</button>
</div>
<!-- Chat Messages for the Active Tab -->
<div id="messages-container">
<!-- Messages are injected here by JavaScript -->
</div>
<!-- Input Area -->
<div id="input-area">
<div id="input-controls">
<label class="agent-toggle">
<input type="checkbox" id="chk-agent-mode">
Agent Mode
</label>
<select id="sel-model">
<!-- Model options injected by JavaScript -->
</select>
</div>
<div id="input-row">
<textarea
id="txt-input"
placeholder="Type your message or agent goal... (Ctrl+Enter to send)"
rows="3"
></textarea>
<button id="btn-send">Send</button>
<button id="btn-stop" style="display:none">Stop</button>
</div>
</div>
</main>
<!-- Right Sidebar: Model and MCP Management -->
<aside id="models-panel">
<div class="panel-header">
<h2>Models</h2>
</div>
<div id="models-list">
<!-- Model cards injected by JavaScript -->
</div>
<div id="add-model-form">
<h3>Add Model</h3>
<input id="inp-model-name"
placeholder="Display name" type="text">
<input id="inp-model-url"
placeholder="Base URL (e.g. http://localhost:11434)"
type="text">
<input id="inp-model-id"
placeholder="Model name (e.g. llama4)"
type="text">
<input id="inp-model-key"
placeholder="API key (blank for local)"
type="password">
<label class="agent-toggle">
<input type="checkbox" id="chk-model-default">
Set as default
</label>
<button id="btn-add-model">Add Model</button>
</div>
<div id="mcp-section">
<div class="panel-header">
<h3>MCP Servers</h3>
</div>
<div id="mcp-list">
<!-- MCP server entries injected by JavaScript -->
</div>
<div id="add-mcp-form">
<input id="inp-mcp-name"
placeholder="Server name" type="text">
<select id="sel-mcp-transport">
<option value="stdio">stdio (local process)</option>
<option value="http">HTTP (remote server)</option>
</select>
<input id="inp-mcp-command"
placeholder="Command (stdio only)" type="text">
<input id="inp-mcp-url"
placeholder="URL (HTTP only)" type="text">
<button id="btn-add-mcp">Add MCP Server</button>
</div>
</div>
</aside>
</div>
<!-- Session Viewer Modal -->
<div id="session-modal" class="modal" style="display:none">
<div class="modal-content">
<div class="modal-header">
<h2 id="modal-title">Session History</h2>
<button id="btn-close-modal">X</button>
</div>
<div id="modal-messages">
<!-- Session messages injected here -->
</div>
</div>
</div>
<script src="/static/app.js"></script>
</body>
</html>
The CSS uses custom properties (variables) for consistent theming throughout the application. The three-panel layout is implemented with CSS Grid, which makes the proportions easy to adjust. The tab spinner animation gives the user clear visual feedback that a task is running in a particular tab.
/* frontend/style.css */
/*
* Application styles for the Agentic Chatbot Orchestrator.
* Uses CSS custom properties for consistent theming.
* The three-panel layout uses CSS Grid.
*/
:root {
--color-bg: #1a1a2e;
--color-surface: #16213e;
--color-surface-2: #0f3460;
--color-accent: #e94560;
--color-accent-2: #533483;
--color-text: #e0e0e0;
--color-text-muted: #888;
--color-border: #2a2a4a;
--color-user-msg: #1e3a5f;
--color-asst-msg: #1a2a1a;
--color-tool-msg: #2a1a3a;
--color-thinking-msg: #2a2a1a;
--radius: 8px;
--font-mono: "Fira Code", "Cascadia Code", monospace;
--font-sans: "Segoe UI", system-ui, sans-serif;
}
* {
box-sizing: border-box;
margin: 0;
padding: 0;
}
body {
background: var(--color-bg);
color: var(--color-text);
font-family: var(--font-sans);
height: 100vh;
overflow: hidden;
}
#app {
display: grid;
grid-template-columns: 240px 1fr 280px;
height: 100vh;
}
aside {
background: var(--color-surface);
border-right: 1px solid var(--color-border);
display: flex;
flex-direction: column;
overflow: hidden;
}
#models-panel {
border-left: 1px solid var(--color-border);
border-right: none;
overflow-y: auto;
}
.panel-header {
display: flex;
align-items: center;
justify-content: space-between;
padding: 12px 16px;
border-bottom: 1px solid var(--color-border);
flex-shrink: 0;
}
.panel-header h2,
.panel-header h3 {
font-size: 14px;
font-weight: 600;
color: var(--color-text-muted);
text-transform: uppercase;
letter-spacing: 0.05em;
}
#sessions-list {
flex: 1;
overflow-y: auto;
padding: 8px;
}
.session-item {
padding: 10px 12px;
border-radius: var(--radius);
cursor: pointer;
margin-bottom: 4px;
border: 1px solid transparent;
transition: background 0.15s;
}
.session-item:hover {
background: var(--color-surface-2);
border-color: var(--color-border);
}
.session-item .session-title {
font-size: 13px;
font-weight: 500;
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
}
.session-item .session-meta {
font-size: 11px;
color: var(--color-text-muted);
margin-top: 2px;
}
.agent-badge {
display: inline-block;
background: var(--color-accent-2);
color: white;
font-size: 10px;
padding: 1px 5px;
border-radius: 3px;
margin-left: 4px;
}
#tab-bar {
display: flex;
align-items: center;
background: var(--color-surface);
border-bottom: 1px solid var(--color-border);
padding: 0 8px;
height: 44px;
flex-shrink: 0;
overflow-x: auto;
}
#tabs-container {
display: flex;
gap: 4px;
flex: 1;
overflow-x: auto;
}
.tab-btn {
display: flex;
align-items: center;
gap: 6px;
padding: 6px 12px;
border-radius: var(--radius) var(--radius) 0 0;
border: 1px solid transparent;
background: transparent;
color: var(--color-text-muted);
cursor: pointer;
font-size: 13px;
white-space: nowrap;
transition: all 0.15s;
}
.tab-btn.active {
background: var(--color-bg);
border-color: var(--color-border);
border-bottom-color: var(--color-bg);
color: var(--color-text);
}
.tab-close {
font-size: 11px;
opacity: 0.5;
padding: 0 2px;
cursor: pointer;
}
.tab-close:hover {
opacity: 1;
color: var(--color-accent);
}
.tab-spinner {
width: 10px;
height: 10px;
border: 2px solid var(--color-accent);
border-top-color: transparent;
border-radius: 50%;
animation: spin 0.8s linear infinite;
flex-shrink: 0;
}
@keyframes spin {
to { transform: rotate(360deg); }
}
#chat-area {
display: flex;
flex-direction: column;
background: var(--color-bg);
overflow: hidden;
}
#messages-container {
flex: 1;
overflow-y: auto;
padding: 16px;
display: flex;
flex-direction: column;
gap: 12px;
}
.message {
max-width: 85%;
padding: 12px 16px;
border-radius: var(--radius);
font-size: 14px;
line-height: 1.6;
}
.message.user {
background: var(--color-user-msg);
align-self: flex-end;
border-bottom-right-radius: 2px;
}
.message.assistant {
background: var(--color-asst-msg);
align-self: flex-start;
border-bottom-left-radius: 2px;
}
.message.tool-call {
background: var(--color-tool-msg);
align-self: flex-start;
font-family: var(--font-mono);
font-size: 12px;
border-left: 3px solid var(--color-accent-2);
max-width: 95%;
}
.message.tool-result {
background: var(--color-tool-msg);
align-self: flex-start;
font-family: var(--font-mono);
font-size: 12px;
border-left: 3px solid #2a5a2a;
max-width: 95%;
}
.message.thinking {
background: var(--color-thinking-msg);
align-self: flex-start;
font-style: italic;
color: var(--color-text-muted);
border-left: 3px solid #5a5a2a;
}
.message-role {
font-size: 11px;
font-weight: 600;
text-transform: uppercase;
letter-spacing: 0.05em;
margin-bottom: 6px;
opacity: 0.7;
}
.message-content {
white-space: pre-wrap;
word-break: break-word;
}
.streaming-cursor::after {
content: "|";
animation: blink 0.7s step-end infinite;
}
@keyframes blink {
50% { opacity: 0; }
}
.agent-iteration {
text-align: center;
font-size: 11px;
color: var(--color-text-muted);
padding: 4px 0;
}
#input-area {
padding: 12px 16px;
border-top: 1px solid var(--color-border);
background: var(--color-surface);
flex-shrink: 0;
}
#input-controls {
display: flex;
align-items: center;
gap: 12px;
margin-bottom: 8px;
}
.agent-toggle {
display: flex;
align-items: center;
gap: 6px;
font-size: 13px;
cursor: pointer;
user-select: none;
white-space: nowrap;
}
#sel-model {
background: var(--color-surface-2);
color: var(--color-text);
border: 1px solid var(--color-border);
border-radius: var(--radius);
padding: 4px 8px;
font-size: 13px;
flex: 1;
min-width: 0;
}
#input-row {
display: flex;
gap: 8px;
align-items: flex-end;
}
#txt-input {
flex: 1;
background: var(--color-surface-2);
color: var(--color-text);
border: 1px solid var(--color-border);
border-radius: var(--radius);
padding: 10px 12px;
font-family: var(--font-sans);
font-size: 14px;
resize: none;
outline: none;
transition: border-color 0.15s;
}
#txt-input:focus {
border-color: var(--color-accent);
}
button {
background: var(--color-accent);
color: white;
border: none;
border-radius: var(--radius);
padding: 8px 16px;
font-size: 13px;
font-weight: 600;
cursor: pointer;
transition: opacity 0.15s;
white-space: nowrap;
}
button:hover { opacity: 0.85; }
button:disabled { opacity: 0.4; cursor: not-allowed; }
#btn-stop { background: #555; }
.model-card {
margin: 8px;
padding: 10px 12px;
background: var(--color-surface-2);
border-radius: var(--radius);
border: 1px solid var(--color-border);
font-size: 12px;
}
.model-card.is-default {
border-color: var(--color-accent);
}
.model-name {
font-weight: 600;
font-size: 13px;
margin-bottom: 4px;
}
.model-detail {
color: var(--color-text-muted);
margin-bottom: 2px;
word-break: break-all;
}
.token-usage {
margin-top: 6px;
padding-top: 6px;
border-top: 1px solid var(--color-border);
display: flex;
gap: 8px;
font-size: 11px;
flex-wrap: wrap;
}
.token-badge {
background: var(--color-bg);
padding: 2px 6px;
border-radius: 4px;
}
.model-actions {
display: flex;
gap: 4px;
margin-top: 6px;
flex-wrap: wrap;
}
.model-actions button {
font-size: 11px;
padding: 3px 8px;
background: var(--color-surface);
}
#add-model-form,
#add-mcp-form {
padding: 12px;
border-top: 1px solid var(--color-border);
display: flex;
flex-direction: column;
gap: 6px;
}
#add-model-form h3 {
font-size: 12px;
color: var(--color-text-muted);
text-transform: uppercase;
margin-bottom: 4px;
}
input[type="text"],
input[type="password"],
select {
background: var(--color-surface-2);
color: var(--color-text);
border: 1px solid var(--color-border);
border-radius: var(--radius);
padding: 7px 10px;
font-size: 12px;
outline: none;
width: 100%;
}
input:focus,
select:focus {
border-color: var(--color-accent);
}
#mcp-section {
border-top: 1px solid var(--color-border);
margin-top: 8px;
}
.mcp-item {
margin: 4px 8px;
padding: 8px 10px;
background: var(--color-surface-2);
border-radius: var(--radius);
font-size: 12px;
display: flex;
align-items: center;
justify-content: space-between;
gap: 8px;
}
.mcp-name {
font-weight: 600;
}
.mcp-transport {
color: var(--color-text-muted);
font-size: 11px;
}
.mcp-item button {
font-size: 11px;
padding: 3px 8px;
background: var(--color-surface);
flex-shrink: 0;
}
.modal {
position: fixed;
inset: 0;
background: rgba(0, 0, 0, 0.7);
display: flex;
align-items: center;
justify-content: center;
z-index: 1000;
}
.modal-content {
background: var(--color-surface);
border: 1px solid var(--color-border);
border-radius: var(--radius);
width: 700px;
max-width: 95vw;
max-height: 80vh;
display: flex;
flex-direction: column;
}
.modal-header {
display: flex;
align-items: center;
justify-content: space-between;
padding: 16px 20px;
border-bottom: 1px solid var(--color-border);
flex-shrink: 0;
}
.modal-header h2 {
font-size: 16px;
font-weight: 600;
}
#modal-messages {
flex: 1;
overflow-y: auto;
padding: 16px 20px;
display: flex;
flex-direction: column;
gap: 10px;
}
::-webkit-scrollbar { width: 6px; }
::-webkit-scrollbar-track { background: transparent; }
::-webkit-scrollbar-thumb {
background: var(--color-border);
border-radius: 3px;
}
The JavaScript application is the most complex part of the frontend. It is organised around a central state object and a set of rendering functions. The state object is the single source of truth. When a WebSocket message arrives, it updates the state and then calls the appropriate render function to update the DOM. This one-way data flow makes the application predictable and easy to debug.
// frontend/app.js
/**
* Agentic Chatbot Orchestrator - Frontend Application
*
* Architecture:
* State -> Render functions -> DOM updates
* Events -> State mutations -> Targeted re-renders
*
* A single WebSocket connection multiplexes all active tabs using
* the tab_id field present in every message. Each tab maintains its
* own message history and running state in the central state object.
*/
"use strict";
// ==================================================================
// CONSTANTS
// ==================================================================
const WS_URL = `ws://${window.location.host}/ws`;
const API_BASE = "/api";
// ==================================================================
// APPLICATION STATE
// ==================================================================
/**
* Central state object. All UI state lives here.
* Mutations must always be followed by the appropriate render call
* to keep the DOM in sync with the state.
*
* Tab object shape:
* {
* tabId: string, - Unique tab identifier
* sessionId: string|null, - Server-side session ID
* title: string, - Display title for the tab button
* isAgentMode: boolean, - Whether agent mode is active
* modelId: string, - ID of the selected LLM model
* messages: Array, - Local message history for rendering
* isRunning: boolean, - Whether a task is currently running
* streamBuffer: Element|null - DOM element for active streaming
* }
*/
const state = {
ws: null,
wsReady: false,
tabs: new Map(), // tabId -> Tab object
activeTabId: null,
tabCounter: 0,
models: [], // All configured LLM models
savedSessions: [], // All saved (closed) sessions
mcpServers: [], // All configured MCP servers
};
// ==================================================================
// WEBSOCKET MANAGEMENT
// ==================================================================
/**
* Initialise the WebSocket connection and register message handlers.
*
* Implements automatic reconnection with a fixed 3-second delay.
* When the connection is re-established, the active tabs remain
* visible in the UI, but their server-side sessions will need to
* be recreated if the server was also restarted.
*/
function initWebSocket() {
const ws = new WebSocket(WS_URL);
state.ws = ws;
ws.onopen = () => {
state.wsReady = true;
console.log("[WS] Connected.");
// Create the first tab once the connection is ready.
// This is called here rather than in init() to ensure the
// WebSocket is open before the first create_session message
// is sent.
if (state.tabs.size === 0) {
createNewTab();
}
};
ws.onmessage = (event) => {
try {
const msg = JSON.parse(event.data);
handleServerMessage(msg);
} catch (err) {
console.error("[WS] Failed to parse message:", err);
}
};
ws.onclose = () => {
state.wsReady = false;
console.warn("[WS] Disconnected. Reconnecting in 3s...");
setTimeout(initWebSocket, 3000);
};
ws.onerror = (err) => {
console.error("[WS] Error:", err);
};
}
/**
* Send a JSON message to the server over the WebSocket.
* Logs a warning and drops the message if the connection is not ready.
*/
function wsSend(data) {
if (state.ws && state.wsReady) {
state.ws.send(JSON.stringify(data));
} else {
console.warn("[WS] Not ready; message dropped:", data);
}
}
// ==================================================================
// SERVER MESSAGE DISPATCHER
// ==================================================================
/**
* Route an incoming server message to the appropriate handler.
* Every message type maps to a specific, focused handler function.
*/
function handleServerMessage(msg) {
const { type, tab_id: tabId } = msg;
switch (type) {
case "session_created": handleSessionCreated(msg); break;
case "stream_start": handleStreamStart(tabId); break;
case "stream_chunk": handleStreamChunk(tabId, msg.content); break;
case "stream_end": handleStreamEnd(tabId, msg); break;
case "stream_cancelled": handleStreamCancelled(tabId); break;
case "agent_start": /* spinner already shown */ break;
case "agent_event": handleAgentEvent(tabId, msg.event); break;
case "agent_complete": handleAgentComplete(tabId); break;
case "agent_cancelled": handleStreamCancelled(tabId); break;
case "tab_closed": handleTabClosed(msg); break;
case "tab_cancelled": setTabRunning(tabId, false); break;
case "error": handleError(tabId, msg.message); break;
default:
console.warn("[WS] Unknown message type:", type);
}
}
// ==================================================================
// TAB MANAGEMENT
// ==================================================================
/**
* Create a new tab and request a corresponding server-side session.
*
* The tab appears immediately in the UI with a placeholder title.
* The session ID is filled in when the server responds with
* 'session_created'. Until then, the tab cannot send messages.
*/
function createNewTab() {
const tabId = `tab-${++state.tabCounter}`;
const modelId = getSelectedModelId();
const tab = {
tabId,
sessionId: null,
title: "New Chat",
isAgentMode: false,
modelId: modelId || "",
messages: [],
isRunning: false,
streamBuffer: null,
};
state.tabs.set(tabId, tab);
state.activeTabId = tabId;
renderTabBar();
renderMessages();
updateInputControls();
// Request a new session from the server.
wsSend({
action: "create_session",
tab_id: tabId,
model_id: modelId || null,
is_agent_mode: false,
});
}
/**
* Switch the active tab to the given tabId and re-render the chat area.
*/
function switchTab(tabId) {
if (!state.tabs.has(tabId)) return;
state.activeTabId = tabId;
renderTabBar();
renderMessages();
updateInputControls();
// Sync the agent mode checkbox and model selector to the new tab.
const tab = state.tabs.get(tabId);
if (tab) {
document.getElementById("chk-agent-mode").checked = tab.isAgentMode;
const sel = document.getElementById("sel-model");
if (tab.modelId) sel.value = tab.modelId;
}
}
/**
* Close a tab, saving its session to the sessions list.
*
* If the closed tab was the active one, the application switches to
* the most recently opened remaining tab. If no tabs remain, a new
* one is created automatically so the user always has a chat window.
*/
function closeTab(tabId) {
const tab = state.tabs.get(tabId);
if (!tab) return;
// Notify the server to cancel any running task and save the session.
if (tab.sessionId) {
wsSend({
action: "close_tab",
tab_id: tabId,
session_id: tab.sessionId,
});
}
state.tabs.delete(tabId);
// Switch to another tab if this was the active one.
if (state.activeTabId === tabId) {
const remaining = [...state.tabs.keys()];
state.activeTabId = remaining.length > 0
? remaining[remaining.length - 1]
: null;
}
if (state.tabs.size === 0) {
// Always maintain at least one tab.
createNewTab();
} else {
renderTabBar();
renderMessages();
updateInputControls();
}
}
/**
* Send the user's message for the currently active tab.
*
* Uses an optimistic update: the user message is rendered immediately
* for instant visual feedback, then sent to the server. The tab is
* marked as running and the input is cleared.
*/
function sendMessage() {
const tab = state.tabs.get(state.activeTabId);
if (!tab || tab.isRunning || !tab.sessionId) return;
const input = document.getElementById("txt-input");
const content = input.value.trim();
if (!content) return;
// Sync agent mode from the checkbox.
tab.isAgentMode = document.getElementById("chk-agent-mode").checked;
tab.modelId = document.getElementById("sel-model").value;
// Optimistic update: show the message immediately.
tab.messages.push({
role: "user",
content,
id: `local-${Date.now()}`,
});
renderMessages();
input.value = "";
input.style.height = "auto";
wsSend({
action: "send_message",
tab_id: tab.tabId,
session_id: tab.sessionId,
content,
model_id: tab.modelId || null,
is_agent_mode: tab.isAgentMode,
});
setTabRunning(tab.tabId, true);
}
/**
* Request cancellation of the running task for the active tab.
*/
function stopCurrentTask() {
const tab = state.tabs.get(state.activeTabId);
if (!tab) return;
wsSend({ action: "cancel_tab", tab_id: tab.tabId });
}
// ==================================================================
// SERVER MESSAGE HANDLERS
// ==================================================================
function handleSessionCreated(msg) {
const tab = state.tabs.get(msg.tab_id);
if (!tab) return;
tab.sessionId = msg.session_id;
tab.modelId = msg.model_id || tab.modelId;
// Sync the model selector if this is the active tab.
if (msg.tab_id === state.activeTabId && tab.modelId) {
document.getElementById("sel-model").value = tab.modelId;
}
}
function handleStreamStart(tabId) {
const tab = state.tabs.get(tabId);
if (!tab) return;
// Create a placeholder message element for the incoming stream.
const msgEl = createMessageElement("assistant", "");
msgEl.querySelector(".message-content")
.classList.add("streaming-cursor");
tab.streamBuffer = msgEl;
if (tabId === state.activeTabId) {
document.getElementById("messages-container").appendChild(msgEl);
scrollToBottom();
}
}
function handleStreamChunk(tabId, content) {
const tab = state.tabs.get(tabId);
if (!tab || !tab.streamBuffer) return;
tab.streamBuffer.querySelector(".message-content").textContent += content;
if (tabId === state.activeTabId) {
scrollToBottom();
}
}
function handleStreamEnd(tabId, msg) {
const tab = state.tabs.get(tabId);
if (!tab) return;
if (tab.streamBuffer) {
const contentEl = tab.streamBuffer
.querySelector(".message-content");
contentEl.classList.remove("streaming-cursor");
// Record the complete message in local state.
tab.messages.push({
role: "assistant",
content: contentEl.textContent,
id: `local-${Date.now()}`,
});
tab.streamBuffer = null;
}
setTabRunning(tabId, false);
// Refresh model list to show updated token counts.
loadModels();
}
/**
* Handle an agent event from the ReACT loop.
*
* Each event type receives a distinct visual treatment so the user
* can follow the agent's reasoning process step by step.
*
* The server sends:
* { "type": "agent_event", "tab_id": "...",
* "event": { "event": "thinking", "data": {...} } }
* So the 'event' parameter here is { event: string, data: object }.
*/
function handleAgentEvent(tabId, event) {
const tab = state.tabs.get(tabId);
if (!tab) return;
// Destructure: event.event is the type string, event.data is the payload.
const eventType = event.event;
const data = event.data;
let msgEl = null;
if (eventType === "iteration") {
if (tabId === state.activeTabId) {
const el = document.createElement("div");
el.className = "agent-iteration";
el.textContent = `Step ${data.current} of ${data.max}`;
document.getElementById("messages-container").appendChild(el);
}
return;
}
if (eventType === "thinking") {
msgEl = createMessageElement("thinking", data.content);
tab.messages.push({
role: "assistant",
content: data.content,
thinking: true,
id: `local-${Date.now()}`,
});
} else if (eventType === "tool_call") {
const argsStr = JSON.stringify(data.arguments, null, 2);
msgEl = createMessageElement(
"tool-call",
`Tool: ${data.tool_name}\nArgs:\n${argsStr}`
);
} else if (eventType === "tool_result") {
const prefix = data.is_error ? "ERROR: " : "Result: ";
msgEl = createMessageElement(
"tool-result",
`${prefix}${data.result}`
);
} else if (eventType === "answer") {
msgEl = createMessageElement("assistant", data.content);
tab.messages.push({
role: "assistant",
content: data.content,
id: `local-${Date.now()}`,
});
} else if (eventType === "error") {
msgEl = createMessageElement(
"tool-result",
`Agent Error: ${data.message}`
);
} else if (eventType === "token_update") {
// Token updates are handled server-side; just refresh the model list.
loadModels();
return;
}
if (msgEl && tabId === state.activeTabId) {
document.getElementById("messages-container").appendChild(msgEl);
scrollToBottom();
}
}
function handleAgentComplete(tabId) {
setTabRunning(tabId, false);
loadModels(); // Refresh token counts.
}
function handleStreamCancelled(tabId) {
const tab = state.tabs.get(tabId);
if (tab && tab.streamBuffer) {
tab.streamBuffer
.querySelector(".message-content")
.classList.remove("streaming-cursor");
tab.streamBuffer = null;
}
setTabRunning(tabId, false);
}
function handleTabClosed(msg) {
// The server has saved the session; refresh the sessions list.
loadSavedSessions();
}
function handleError(tabId, message) {
const tab = state.tabs.get(tabId);
if (!tab) return;
const msgEl = createMessageElement(
"tool-result",
`Error: ${message}`
);
if (tabId === state.activeTabId) {
document.getElementById("messages-container").appendChild(msgEl);
scrollToBottom();
}
setTabRunning(tabId, false);
}
// ==================================================================
// RENDERING FUNCTIONS
// ==================================================================
/**
* Render the tab bar from the current state.
*
* Each tab gets a button showing a spinner when running, an optional
* agent-mode indicator, the tab title, and a close button.
*/
function renderTabBar() {
const container = document.getElementById("tabs-container");
container.innerHTML = "";
for (const [tabId, tab] of state.tabs) {
const isActive = tabId === state.activeTabId;
const btn = document.createElement("button");
btn.className = "tab-btn" + (isActive ? " active" : "");
btn.dataset.tabId = tabId;
const agentPrefix = tab.isAgentMode ? "[A] " : "";
const titleText = agentPrefix + tab.title;
// Build the tab button content safely to avoid XSS.
if (tab.isRunning) {
const spinner = document.createElement("span");
spinner.className = "tab-spinner";
btn.appendChild(spinner);
}
const titleSpan = document.createElement("span");
titleSpan.className = "tab-title";
titleSpan.textContent = titleText;
const closeSpan = document.createElement("span");
closeSpan.className = "tab-close";
closeSpan.textContent = "x";
closeSpan.dataset.closeTabId = tabId;
btn.appendChild(titleSpan);
btn.appendChild(closeSpan);
btn.addEventListener("click", (e) => {
// Check if the close button was clicked.
const closeId = e.target.dataset.closeTabId
|| e.target.closest("[data-close-tab-id]")
?.dataset.closeTabId;
if (closeId) {
closeTab(closeId);
} else {
switchTab(tabId);
}
});
container.appendChild(btn);
}
}
/**
* Render all messages for the currently active tab.
*
* Clears and rebuilds the messages container from the tab's local
* message array. Called on tab switch and initial load.
*/
function renderMessages() {
const container = document.getElementById("messages-container");
container.innerHTML = "";
if (!state.activeTabId) {
const placeholder = document.createElement("div");
placeholder.style.cssText =
"text-align:center;color:#888;margin-top:40px;font-size:14px";
placeholder.textContent = "Open a new tab to start chatting.";
container.appendChild(placeholder);
return;
}
const tab = state.tabs.get(state.activeTabId);
if (!tab) return;
for (const msg of tab.messages) {
const cssClass = msg.thinking ? "thinking" : msg.role;
container.appendChild(
createMessageElement(cssClass, msg.content)
);
}
// Re-attach the streaming buffer if this tab is still streaming.
if (tab.streamBuffer) {
container.appendChild(tab.streamBuffer);
}
scrollToBottom();
}
/**
* Render the model list in the right sidebar and populate the
* model selector in the input area.
*/
function renderModels() {
const listContainer = document.getElementById("models-list");
listContainer.innerHTML = "";
const selector = document.getElementById("sel-model");
const currentValue = selector.value;
selector.innerHTML = "";
for (const model of state.models) {
// Build the model card safely without innerHTML injection.
const card = document.createElement("div");
card.className = "model-card" + (model.is_default ? " is-default" : "");
const nameDiv = document.createElement("div");
nameDiv.className = "model-name";
nameDiv.textContent = model.display_name
+ (model.is_default ? " \u2605" : "");
const modelIdDiv = document.createElement("div");
modelIdDiv.className = "model-detail";
modelIdDiv.textContent = model.model_name;
const urlDiv = document.createElement("div");
urlDiv.className = "model-detail";
urlDiv.style.fontSize = "11px";
urlDiv.style.color = "#666";
urlDiv.textContent = model.base_url;
const tokenDiv = document.createElement("div");
tokenDiv.className = "token-usage";
tokenDiv.innerHTML =
`<span class="token-badge">In: ${formatTokens(model.input_tokens_used)}</span>` +
`<span class="token-badge">Out: ${formatTokens(model.output_tokens_used)}</span>`;
const actionsDiv = document.createElement("div");
actionsDiv.className = "model-actions";
if (!model.is_default) {
const setDefaultBtn = document.createElement("button");
setDefaultBtn.textContent = "Set Default";
setDefaultBtn.addEventListener("click", () =>
setDefaultModel(model.id)
);
actionsDiv.appendChild(setDefaultBtn);
}
const deleteBtn = document.createElement("button");
deleteBtn.textContent = "Delete";
deleteBtn.addEventListener("click", () => deleteModel(model.id));
actionsDiv.appendChild(deleteBtn);
card.appendChild(nameDiv);
card.appendChild(modelIdDiv);
card.appendChild(urlDiv);
card.appendChild(tokenDiv);
card.appendChild(actionsDiv);
listContainer.appendChild(card);
// Add option to the model selector.
const option = document.createElement("option");
option.value = model.id;
option.textContent = model.display_name;
selector.appendChild(option);
}
// Restore the previously selected value if it still exists.
if (currentValue) selector.value = currentValue;
// If nothing is selected, pick the default model.
if (!selector.value && state.models.length > 0) {
const def = state.models.find(m => m.is_default);
if (def) selector.value = def.id;
}
}
/**
* Render the saved sessions list in the left sidebar.
* Clicking a session opens the session viewer modal.
*/
function renderSavedSessions() {
const container = document.getElementById("sessions-list");
container.innerHTML = "";
if (state.savedSessions.length === 0) {
const empty = document.createElement("div");
empty.style.cssText = "padding:16px;color:#888;font-size:13px";
empty.textContent = "No saved sessions yet.";
container.appendChild(empty);
return;
}
for (const session of state.savedSessions) {
const item = document.createElement("div");
item.className = "session-item";
const titleDiv = document.createElement("div");
titleDiv.className = "session-title";
titleDiv.textContent = session.title;
if (session.is_agent_mode) {
const badge = document.createElement("span");
badge.className = "agent-badge";
badge.textContent = "Agent";
titleDiv.appendChild(badge);
}
const metaDiv = document.createElement("div");
metaDiv.className = "session-meta";
const date = new Date(session.updated_at).toLocaleDateString();
metaDiv.textContent = `${session.message_count} messages \u2022 ${date}`;
item.appendChild(titleDiv);
item.appendChild(metaDiv);
item.addEventListener("click", () =>
openSessionModal(session.id)
);
container.appendChild(item);
}
}
/**
* Render the MCP servers list in the right sidebar.
*/
function renderMCPServers() {
const container = document.getElementById("mcp-list");
container.innerHTML = "";
for (const server of state.mcpServers) {
const item = document.createElement("div");
item.className = "mcp-item";
const infoDiv = document.createElement("div");
const nameDiv = document.createElement("div");
nameDiv.className = "mcp-name";
nameDiv.textContent = server.name;
const transportDiv = document.createElement("div");
transportDiv.className = "mcp-transport";
transportDiv.textContent = server.transport;
infoDiv.appendChild(nameDiv);
infoDiv.appendChild(transportDiv);
const removeBtn = document.createElement("button");
removeBtn.textContent = "Remove";
removeBtn.addEventListener("click", () =>
deleteMCPServer(server.id)
);
item.appendChild(infoDiv);
item.appendChild(removeBtn);
container.appendChild(item);
}
}
/**
* Create a styled message DOM element.
*
* The cssClass parameter maps to CSS classes defined in style.css.
* Content is set via textContent (not innerHTML) to prevent XSS.
*/
function createMessageElement(cssClass, content) {
const roleLabels = {
user: "You",
assistant: "Assistant",
"tool-call": "Tool Call",
"tool-result": "Tool Result",
thinking: "Reasoning",
tool: "Tool Result",
};
const el = document.createElement("div");
el.className = `message ${cssClass}`;
const roleEl = document.createElement("div");
roleEl.className = "message-role";
roleEl.textContent = roleLabels[cssClass] || cssClass;
const contentEl = document.createElement("div");
contentEl.className = "message-content";
contentEl.textContent = content;
el.appendChild(roleEl);
el.appendChild(contentEl);
return el;
}
// ==================================================================
// SESSION VIEWER MODAL
// ==================================================================
/**
* Open the session history modal for a saved session.
*
* Fetches the full session detail from the REST API and renders all
* messages in the modal. System messages are hidden from the user
* since they contain internal agent instructions.
*/
async function openSessionModal(sessionId) {
const modal = document.getElementById("session-modal");
const titleEl = document.getElementById("modal-title");
const messagesEl = document.getElementById("modal-messages");
messagesEl.innerHTML = "";
const loading = document.createElement("div");
loading.style.color = "#888";
loading.textContent = "Loading...";
messagesEl.appendChild(loading);
modal.style.display = "flex";
try {
const response = await fetch(`${API_BASE}/sessions/${sessionId}`);
if (!response.ok) throw new Error(`HTTP ${response.status}`);
const session = await response.json();
titleEl.textContent = session.title || "Session History";
messagesEl.innerHTML = "";
for (const msg of session.messages) {
if (msg.role === "system") continue;
const cssClass = msg.role === "tool" ? "tool-result" : msg.role;
messagesEl.appendChild(
createMessageElement(cssClass, msg.content)
);
}
} catch (err) {
messagesEl.innerHTML = "";
const errEl = document.createElement("div");
errEl.style.color = "red";
errEl.textContent = `Failed to load session: ${err.message}`;
messagesEl.appendChild(errEl);
}
}
function closeSessionModal() {
document.getElementById("session-modal").style.display = "none";
}
// ==================================================================
// API CALLS
// ==================================================================
async function loadModels() {
try {
const res = await fetch(`${API_BASE}/models`);
if (!res.ok) throw new Error(`HTTP ${res.status}`);
state.models = await res.json();
renderModels();
} catch (err) {
console.error("Failed to load models:", err);
}
}
async function loadSavedSessions() {
try {
const res = await fetch(`${API_BASE}/sessions`);
if (!res.ok) throw new Error(`HTTP ${res.status}`);
state.savedSessions = await res.json();
renderSavedSessions();
} catch (err) {
console.error("Failed to load sessions:", err);
}
}
async function loadMCPServers() {
try {
const res = await fetch(`${API_BASE}/mcp-servers`);
if (!res.ok) throw new Error(`HTTP ${res.status}`);
state.mcpServers = await res.json();
renderMCPServers();
} catch (err) {
console.error("Failed to load MCP servers:", err);
}
}
async function addModel() {
const name = document.getElementById("inp-model-name").value.trim();
const url = document.getElementById("inp-model-url").value.trim();
const modelId = document.getElementById("inp-model-id").value.trim();
const key = document.getElementById("inp-model-key").value.trim();
const isDef = document.getElementById("chk-model-default").checked;
if (!name || !url || !modelId) {
alert("Please fill in the display name, base URL, and model name.");
return;
}
try {
const res = await fetch(`${API_BASE}/models`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
display_name: name,
base_url: url,
model_name: modelId,
api_key: key,
is_default: isDef,
}),
});
if (!res.ok) throw new Error(`HTTP ${res.status}`);
// Clear the form fields.
document.getElementById("inp-model-name").value = "";
document.getElementById("inp-model-url").value = "";
document.getElementById("inp-model-id").value = "";
document.getElementById("inp-model-key").value = "";
document.getElementById("chk-model-default").checked = false;
await loadModels();
} catch (err) {
alert(`Failed to add model: ${err.message}`);
}
}
async function deleteModel(modelId) {
if (!confirm("Delete this model configuration?")) return;
try {
await fetch(`${API_BASE}/models/${modelId}`, { method: "DELETE" });
await loadModels();
} catch (err) {
alert(`Failed to delete model: ${err.message}`);
}
}
async function setDefaultModel(modelId) {
try {
await fetch(`${API_BASE}/models/${modelId}/set-default`, {
method: "PUT",
});
await loadModels();
} catch (err) {
alert(`Failed to set default model: ${err.message}`);
}
}
async function addMCPServer() {
const name = document.getElementById("inp-mcp-name").value.trim();
const transport = document.getElementById("sel-mcp-transport").value;
const command = document.getElementById("inp-mcp-command").value.trim();
const url = document.getElementById("inp-mcp-url").value.trim();
if (!name) {
alert("Please enter a server name.");
return;
}
try {
const res = await fetch(`${API_BASE}/mcp-servers`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
name,
transport,
command: command || null,
url: url || null,
}),
});
if (!res.ok) throw new Error(`HTTP ${res.status}`);
document.getElementById("inp-mcp-name").value = "";
document.getElementById("inp-mcp-command").value = "";
document.getElementById("inp-mcp-url").value = "";
await loadMCPServers();
} catch (err) {
alert(`Failed to add MCP server: ${err.message}`);
}
}
async function deleteMCPServer(serverId) {
if (!confirm("Remove this MCP server?")) return;
try {
await fetch(`${API_BASE}/mcp-servers/${serverId}`, {
method: "DELETE",
});
await loadMCPServers();
} catch (err) {
alert(`Failed to remove MCP server: ${err.message}`);
}
}
async function clearAllSessions() {
if (!confirm(
"Delete ALL sessions? This cannot be undone."
)) return;
try {
await fetch(`${API_BASE}/sessions`, { method: "DELETE" });
state.savedSessions = [];
renderSavedSessions();
} catch (err) {
alert(`Failed to clear sessions: ${err.message}`);
}
}
// ==================================================================
// UTILITY FUNCTIONS
// ==================================================================
/**
* Update the running state for a tab and re-render the tab bar
* and input controls to reflect the change.
*/
function setTabRunning(tabId, isRunning) {
const tab = state.tabs.get(tabId);
if (!tab) return;
tab.isRunning = isRunning;
renderTabBar();
if (tabId === state.activeTabId) {
updateInputControls();
}
}
/**
* Update the send/stop button visibility and input enabled state
* based on whether the active tab has a running task.
*/
function updateInputControls() {
const tab = state.tabs.get(state.activeTabId);
const sendBtn = document.getElementById("btn-send");
const stopBtn = document.getElementById("btn-stop");
const input = document.getElementById("txt-input");
if (!tab || !tab.sessionId) {
// Tab exists but session not yet created; disable input.
sendBtn.disabled = true;
stopBtn.style.display = "none";
input.disabled = true;
return;
}
if (tab.isRunning) {
sendBtn.style.display = "none";
stopBtn.style.display = "block";
input.disabled = true;
} else {
sendBtn.style.display = "block";
sendBtn.disabled = false;
stopBtn.style.display = "none";
input.disabled = false;
input.focus();
}
}
/** Return the currently selected model ID from the selector. */
function getSelectedModelId() {
const sel = document.getElementById("sel-model");
return sel ? sel.value : "";
}
/** Scroll the messages container to the bottom. */
function scrollToBottom() {
const container = document.getElementById("messages-container");
container.scrollTop = container.scrollHeight;
}
/**
* Format a token count for compact display.
* Values >= 1M are shown as "1.2M", >= 1K as "1.2K", otherwise as-is.
*/
function formatTokens(count) {
if (count >= 1_000_000) return (count / 1_000_000).toFixed(1) + "M";
if (count >= 1_000) return (count / 1_000).toFixed(1) + "K";
return String(count);
}
// ==================================================================
// EVENT LISTENERS
// ==================================================================
/**
* Wire up all static UI event listeners.
* Called once after the DOM is ready.
*/
function initEventListeners() {
document.getElementById("btn-new-tab")
.addEventListener("click", createNewTab);
document.getElementById("btn-send")
.addEventListener("click", sendMessage);
document.getElementById("btn-stop")
.addEventListener("click", stopCurrentTask);
document.getElementById("btn-add-model")
.addEventListener("click", addModel);
document.getElementById("btn-add-mcp")
.addEventListener("click", addMCPServer);
document.getElementById("btn-clear-all")
.addEventListener("click", clearAllSessions);
document.getElementById("btn-close-modal")
.addEventListener("click", closeSessionModal);
// Ctrl+Enter or Shift+Enter sends the message.
document.getElementById("txt-input")
.addEventListener("keydown", (e) => {
if (e.key === "Enter" && (e.ctrlKey || e.shiftKey)) {
e.preventDefault();
sendMessage();
}
});
// Auto-resize the textarea as the user types, up to 200px.
document.getElementById("txt-input")
.addEventListener("input", function () {
this.style.height = "auto";
this.style.height =
Math.min(this.scrollHeight, 200) + "px";
});
// Close the modal when clicking on the backdrop.
document.getElementById("session-modal")
.addEventListener("click", (e) => {
if (e.target.id === "session-modal") {
closeSessionModal();
}
});
// Update the active tab's model when the selector changes.
document.getElementById("sel-model")
.addEventListener("change", (e) => {
const tab = state.tabs.get(state.activeTabId);
if (tab) tab.modelId = e.target.value;
});
// Sync the agent mode checkbox to the active tab's state.
document.getElementById("chk-agent-mode")
.addEventListener("change", (e) => {
const tab = state.tabs.get(state.activeTabId);
if (tab) tab.isAgentMode = e.target.checked;
});
}
// ==================================================================
// APPLICATION ENTRY POINT
// ==================================================================
/**
* Initialise the application.
*
* Loads initial data from the REST API, wires up event listeners,
* and opens the WebSocket connection. The first tab is created inside
* the WebSocket onopen handler to guarantee the connection is ready
* before the create_session message is sent.
*/
async function init() {
initEventListeners();
// Load configuration data before opening the WebSocket so that
// the model selector is populated when the first tab is created.
await Promise.all([
loadModels(),
loadSavedSessions(),
loadMCPServers(),
]);
// Open the WebSocket. The first tab is created in ws.onopen.
initWebSocket();
}
document.addEventListener("DOMContentLoaded", init);
CHAPTER 12: RUNNING THE APPLICATION
With all files in place, the application is ready to run. This section walks through the complete process from a fresh checkout to a running application.
STEP 1: VERIFY THE DIRECTORY STRUCTURE
Before starting, confirm that all files are in the correct locations:
agentic_chatbot/
backend/
__init__.py (empty)
config.py
database.py
main.py
orchestrator.py
models/
__init__.py (empty)
agent.py
llm_model.py
mcp_server.py
session.py
services/
__init__.py (empty)
agent_service.py
llm_service.py
mcp_service.py
session_service.py
frontend/
app.js
index.html
style.css
requirements.txt
STEP 2: CREATE AND ACTIVATE THE VIRTUAL ENVIRONMENT
cd agentic_chatbot
python -m venv venv
On Linux and macOS:
source venv/bin/activate
On Windows:
venv\Scripts\activate
STEP 3: INSTALL DEPENDENCIES
pip install -r requirements.txt
This installs FastAPI, uvicorn, httpx, aiosqlite, pydantic, websockets, aiofiles, and FastMCP 3. FastMCP 3 is the MCP client library that manages all MCP server connections.
STEP 4: START A LOCAL LLM SERVER (OPTIONAL BUT RECOMMENDED)
If you have Ollama installed, start it and pull a model:
ollama serve
In a separate terminal:
ollama pull llama4
If you prefer LM Studio, open it, download a model, and start the local server from the Developer tab. LM Studio's server runs on port 1234 by default.
If you are using only remote models (OpenAI, Mistral, etc.), you can skip this step and configure the model through the UI after starting the application.
STEP 5: START THE BACKEND SERVER
From the agentic_chatbot directory (the project root, not the backend directory):
uvicorn backend.main:app --host 0.0.0.0 --port 8000 --reload
The --reload flag enables hot reloading during development. The --host 0.0.0.0 flag makes the server accessible from other machines on the network, which is useful for testing from a phone or tablet.
You should see output similar to:
INFO: Started server process [12345]
INFO: Waiting for application startup.
INFO: Starting Agentic Chatbot Orchestrator v1.0.0
INFO: Database connected: chatbot.db
INFO: Seeded default LLM model.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000
STEP 6: OPEN THE APPLICATION
Open your browser and navigate to:
http://localhost:8000
The application loads immediately. A first tab is created automatically. The default model (Ollama with llama4) is pre-configured. If Ollama is running with llama4 pulled, you can start chatting immediately by typing in the input box and pressing Ctrl+Enter or clicking Send.
ADDING A REMOTE MODEL (OPENAI EXAMPLE)
To add OpenAI's GPT-4o, fill in the Add Model form in the right sidebar:
Display name: OpenAI GPT-4o
Base URL: https://api.openai.com
Model name: gpt-4o
API key: sk-your-openai-api-key-here
Click Add Model. The model appears in the list immediately. To make it the default, click Set Default on its card.
ADDING A LOCAL LM STUDIO MODEL
LM Studio's local server runs on port 1234 by default:
Display name: LM Studio (Mistral)
Base URL: http://localhost:1234
Model name: mistral-7b-instruct
API key: (leave blank)
ADDING AN MCP SERVER
FastMCP 3 connects to MCP servers using the same configuration fields as before. The MCP filesystem server is a good first MCP server to try. Install it with Node.js:
npm install -g @modelcontextprotocol/server-filesystem
Then add it in the application's MCP Servers section:
Server name: Filesystem
Transport: stdio (local process)
Command: npx
Args: @modelcontextprotocol/server-filesystem /path/to/directory
FastMCP 3 will launch the server subprocess automatically on first use, complete the MCP handshake, and cache the tool list. No manual subprocess management is required.
For a remote HTTP MCP server, set the transport to HTTP and provide the server's URL. FastMCP 3 handles the HTTP session management automatically.
TESTING AGENT MODE
Open a new tab by clicking the + button. Check the Agent Mode checkbox. Type a goal such as:
Search for information about the ReACT paper and summarise its key contributions.
Click Send. If you have an MCP server with web search capabilities configured, the agent will use it. Without tools, the agent will still reason through the problem using its training knowledge and provide a structured answer.
CHECKING THE HEALTH ENDPOINT
The health check endpoint shows the connectivity status of all configured models:
http://localhost:8000/api/health
This is useful for diagnosing connectivity issues with local or remote LLM servers.
CHAPTER 13: WORKING WITH MULTIPLE CONCURRENT AGENTS
The real power of this application becomes apparent when you open multiple tabs and start agents running simultaneously. Open three tabs by clicking the + button three times. In the first tab, type a simple question in chat mode and send it. In the second tab, enable agent mode and give it a research task. In the third tab, enable agent mode and give it a different task. All three run concurrently.
You can watch the tab spinners animate independently as each tab's task makes progress. Switching between tabs shows you the current state of each conversation. The agents do not interfere with each other because each runs in its own asyncio Task with its own session state and its own message history.
The orchestrator handles the coordination transparently. From the user's perspective, it simply works: multiple things happen at once, each in its own tab, each with its own history and its own model.
A key architectural point is that the asyncio event loop is single-threaded. All the concurrent tasks run on the same thread, interleaved by the event loop. This is safe because all the operations that could block (LLM API calls, MCP tool calls via FastMCP 3, database writes) are async and yield control to the event loop while waiting. The event loop then runs another task until it too yields. This cooperative multitasking is extremely efficient for I/O-bound workloads like LLM applications.
If you needed CPU-bound parallelism (for example, running local model inference in-process), you would use asyncio.run_in_executor() to offload the work to a thread pool. But since this application delegates all model inference to external processes (Ollama, LM Studio, remote APIs), the asyncio approach is perfectly sufficient.
You can verify concurrent execution by opening the browser's developer tools and watching the WebSocket messages. You will see messages for different tab_ids interleaved with each other, confirming that the server is genuinely processing them concurrently.
CHAPTER 14: EXTENDING THE APPLICATION
The clean architecture makes the application easy to extend. Here are the most valuable extensions to consider.
Adding streaming to the agent loop would improve the user experience significantly. Currently, the agent uses non-streaming LLM calls because it needs the complete response to parse tool calls. However, many LLM APIs support streaming tool calls where the tool call arguments are streamed incrementally. Implementing this would require parsing the streaming response for tool call deltas, which is more complex but would make the agent feel more responsive.
Adding support for multi-modal models (those that accept images) would require extending the Message model to include image content items and updating the LLM service to send them in the correct format. The OpenAI API supports this through the content array format where each item can be either text or an image URL or base64-encoded image data.
Adding a RAG (Retrieval-Augmented Generation) capability would involve creating an MCP server that wraps a vector database. The agent would call this server's search tool to retrieve relevant documents before answering questions. This is a powerful pattern for building knowledge bases over proprietary documents. With FastMCP 3, adding a new MCP server is a matter of configuration: no code changes are needed in the application itself.
Implementing proper token counting for streaming responses would improve the accuracy of the token usage statistics. Currently, the streaming path estimates token counts based on word count. The exact approach would be to use a tokenizer library like tiktoken for OpenAI models or the appropriate tokenizer for other models.
Adding user authentication would be important for a production deployment. FastAPI integrates cleanly with OAuth2 and JWT tokens. Each user would have their own model configurations, sessions, and agent definitions.
The following example shows how the asyncio.gather pattern enables concurrent health checks, illustrating the broader principle of running multiple async operations simultaneously:
# Example: concurrent health check pattern (already in main.py)
# This snippet illustrates the asyncio.gather pattern for
# running multiple async operations concurrently.
async def check_all_models_concurrently(models: list) -> dict:
"""
Check the reachability of all models at the same time.
asyncio.gather runs all the check_model coroutines concurrently.
The total time is the time of the slowest single check, not the
sum of all checks. For 5 models each taking 2 seconds, gather
takes ~2 seconds instead of ~10 seconds.
"""
async def check_one(model) -> tuple[str, bool]:
try:
async with httpx.AsyncClient(timeout=5.0) as client:
resp = await client.get(
f"{model.base_url.rstrip('/')}/v1/models"
)
return model.id, resp.status_code < 400
except Exception:
return model.id, False
results = await asyncio.gather(
*[check_one(m) for m in models],
return_exceptions=True,
)
return {
model_id: is_healthy
for result in results
if isinstance(result, tuple)
for model_id, is_healthy in [result]
}
CHAPTER 15: PRODUCTION CONSIDERATIONS
Deploying this application to production requires attention to several areas that development mode glosses over.
The WebSocket connection should be secured with TLS (wss:// instead of ws://) in production. This is typically handled by a reverse proxy like Nginx or Caddy that terminates TLS and forwards to the FastAPI backend. The CORS configuration should be tightened to only allow the specific domain where the frontend is served.
The SQLite database is appropriate for single-user or small-team use. For larger deployments, migrating to PostgreSQL with asyncpg would provide better concurrent write performance. The database layer is designed with this migration in mind: all SQL is in the Database class, so changing the backend requires only changing that class.
FastMCP 3 manages MCP server subprocess lifecycles automatically. When the application shuts down cleanly via the lifespan manager, FastMCP 3's client shutdown method terminates all stdio subprocesses gracefully. A process supervisor like systemd or supervisord should still be used to ensure the application itself restarts cleanly after a crash.
Rate limiting on the WebSocket endpoint prevents a single user from overwhelming the backend with concurrent requests. FastAPI's middleware system makes it straightforward to add rate limiting using libraries like slowapi.
Logging should be configured to write to a file or a log aggregation service in production. The application already uses Python's standard logging module throughout, so adding a file handler or a service like Datadog or Elasticsearch is a matter of configuration.
The following Nginx configuration handles both regular HTTP and WebSocket upgrade requests, with appropriate timeouts for long-running LLM responses:
# /etc/nginx/sites-available/agentic-chatbot
# Nginx reverse proxy configuration for the Agentic Chatbot.
# Handles HTTP, HTTPS, and WebSocket upgrade requests.
server {
listen 443 ssl http2;
server_name your-domain.com;
ssl_certificate /etc/letsencrypt/live/your-domain.com/fullchain.pem;
ssl_certificate_key /etc/letsencrypt/live/your-domain.com/privkey.pem;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_http_version 1.1;
# These headers are required for WebSocket upgrade to work.
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
# Long timeouts are essential for LLM streaming responses.
# A complex agent task can take several minutes.
proxy_read_timeout 300s;
proxy_send_timeout 300s;
}
}
server {
listen 80;
server_name your-domain.com;
return 301 https://$host$request_uri;
}
For running the application as a systemd service on Linux, create the following service file:
# /etc/systemd/system/agentic-chatbot.service
# systemd service definition for the Agentic Chatbot Orchestrator.
[Unit]
Description=Agentic Chatbot Orchestrator
After=network.target
[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/agentic_chatbot
ExecStart=/opt/agentic_chatbot/venv/bin/uvicorn \
backend.main:app \
--host 127.0.0.1 \
--port 8000 \
--workers 1
Restart=on-failure
RestartSec=5
[Install]
WantedBy=multi-user.target
Note that workers must be set to 1 when using asyncio-based in-memory state (the active sessions dictionary and the orchestrator's task registry). Multiple workers would each have their own independent copies of these data structures, causing sessions to be lost when requests are routed to a different worker. For multi-worker deployments, the active session state would need to move to a shared store like Redis.
Enable and start the service with:
sudo systemctl enable agentic-chatbot
sudo systemctl start agentic-chatbot
sudo systemctl status agentic-chatbot
CONCLUSION
You have now seen the complete architecture and implementation of a production-quality Agentic AI Chatbot Orchestrator. The application demonstrates several important principles that apply far beyond this specific project.
Clean architecture pays dividends from the very first day. By separating data models, business logic, orchestration, and API concerns into distinct layers, every part of the codebase is easy to understand in isolation. When you need to change how LLM communication works, you change the LLM service. When you need to change how sessions are stored, you change the database layer. Nothing else needs to touch.
Async-first design is essential for AI applications. LLM calls are slow by nature: a complex agent task might take 30 seconds or more. If the application were synchronous, every user would have to wait for every other user's requests to complete. With asyncio, hundreds of concurrent sessions can run on a single server, each making progress whenever their I/O operations complete.
The ReACT pattern is a powerful and surprisingly simple foundation for autonomous agents. The loop of reasoning, acting, and observing is intuitive to understand, easy to implement, and effective for a wide range of tasks. The key is giving the agent good tools (via MCP servers managed by FastMCP 3) and a clear system prompt that explains how to use them.
The OpenAI API compatibility layer is what makes this application work with both local and remote models without any code changes. By targeting a single, widely-adopted interface, the application gains access to the entire ecosystem of LLM servers, from a laptop running Ollama with llama4 to a cloud-hosted GPT-4o endpoint.
Finally, the user experience principle of making everything visible and controllable is what separates a useful tool from a frustrating black box. The user can see what the agent is thinking, what tools it is calling, and what results it is getting. They can stop a task at any time, switch between concurrent sessions, and review the history of any past conversation. This transparency builds trust and makes the application genuinely useful for real work.
The code in this tutorial is a complete, working foundation. Take it, run it, and build on it. The architecture is designed to grow with your needs.
