When you spin up a Claude‑powered assistant that can read files, browse the web, or edit code, a single terminal window quickly becomes a black‑box. You lose sight of which agent is speaking, you can’t cancel a runaway tool‑use request, and the UI freezes the moment a network hiccup hits. The only sane way to tame dozens of concurrent Claude instances is a real‑time dashboard that keeps every conversation alive, streams responses, and isolates failures.
In this guide we build exactly that: a desktop GUI where you can launch, monitor, and interact with **multiple Claude agents in parallel**. You’ll see how to juggle async HTTP calls, thread pools, and the new Computer Use API without ever blocking the main UI thread. By the end you’ll have a production‑grade app that can run ten + agents simultaneously, each with its own model (Sonnet, Opus, etc.), memory store, and rate‑limit budget.
- Use Anthropic SDK v1.18+ with Python 3.12 and a thread‑pool executor to keep the UI responsive.
- Separate session objects store per‑agent context, model, and quota counters.
- Wrap every API call with `tenacity` retry + circuit‑breaker logic to sandbox rate‑limit failures.
- Stream Claude’s `messages` response into a QTextEdit widget using a tiny asyncio‑to‑Qt bridge.
- Package the app with PyInstaller or Tauri 2.0 for cross‑platform desktop distribution.
Before you start: Python 3.12, Anthropic SDK v1.18+, PyQt 6.6 (or Tauri 2.0 + RustPython), `tenacity`, `pydantic`, `SQLAlchemy`, `aiohttp`, an Anthropic API key with access to Claude 3.5 Sonnet and Claude 4.0 Opus, and a recent SQLite3 build for local state persistence.
What is a Multi‑Claude Agent Dashboard?
A centralized dashboard for multiple Claude Agent instances is a custom GUI application built using a framework like PyQt or Tauri. It manages separate API sessions, conversation contexts, and UI panels for each agent. The core challenge is architecting a non‑blocking event loop or thread pool to run agents in parallel while streaming their responses live to the interface.
System Architecture & Workflow
The dashboard sits in a classic **client‑side** pattern:
- **UI Layer (PyQt)** – renders a list of agents, individual chat panes, and global controls.
- **Execution Layer** – a `ThreadPoolExecutor` that runs a thin wrapper around the Anthropic client for each agent.
- **State Layer** – SQLite + SQLAlchemy stores per‑agent memory, last message IDs, and quota counters.
- **Network Layer** – `aiohttp` async session inside the wrapper streams Claude’s `messages` endpoint.
The data flow is simple: UI → Executor → Async Claude call → Stream → UI. A global **Quota Broker** watches every request, queues excess traffic, and injects back‑off when Anthropic returns `429` or `429‑quota`.
graph TD
UI[UI (PyQt)] -->|submit request| EXEC[ThreadPoolExecutor]
EXEC -->|async call| CLAUDE[Claude API (aiohttp)]
CLAUDE -->|stream chunks| UI
UI -->|state ops| DB[SQLite via SQLAlchemy]
EXEC -->|quota check| QUOTA[Quota Broker]
QUOTA -->|allow/hold| CLAUDE
—
The 2026 Tool Stack: What You’ll Need to Build
| Component | Why it matters in 2026 | Recommended version |
|---|---|---|
| **Anthropic SDK** | First‑class Python client; supports streaming, Computer Use, and dynamic rate‑limit headers. | `anthropic==1.18.4` |
| **GUI framework** | PyQt 6.6 gives native desktop windows and full thread support; Tauri 2.0 + RustPython is an alternative for a slimmer binary. | `PyQt6==6.6.1` or `tauri==2.0.0` |
| **Async & concurrency** | `asyncio` for streaming; `concurrent.futures.ThreadPoolExecutor` to keep UI snappy. | Python 3.12 |
| **State & logging** | `SQLAlchemy` + SQLite for durable memory, `loguru` for easy structured logs. | `SQLAlchemy==2.0.30`, `loguru==0.7.2` |
| **Retry / circuit‑breaker** | Tenacity implements exponential backoff and jitter, essential for Anthropic’s per‑model quotas. | `tenacity==9.0.0` |
| **Data validation** | Pydantic validates payloads both inbound (Claude) and outbound (tool results). | `pydantic==2.7.0` |
**My take:** I start with PyQt for the UI because its signal/slot mechanism plays nicely with a thread pool. When I need a truly zero‑runtime‑dependency binary (e.g., for corporate air‑gapped machines) I switch to Tauri + RustPython, but the core concurrency model stays the same.
—
Core Architecture: Dashboard Design Patterns
Single‑Threaded vs. Multi‑Threaded vs. Async Event Loop
| Model | Pros | Cons |
|---|---|---|
| **Single‑threaded (Qt event loop only)** | Simpler code; no cross‑thread marshaling. | Any blocking `await` stalls the whole UI; unacceptable for >2 agents. |
| **Multi‑threaded (ThreadPoolExecutor + sync wrapper)** | UI stays fluid; each agent runs isolated. | Python GIL limits CPU‑bound work; but Claude calls are I/O‑bound, so threads are fine. |
| **Pure asyncio (Qt + QEventLoop integration)** | Minimal threads; elegant streaming. | Qt‑async integration is still quirky; you need `qasync` or custom bridge, increasing complexity. |
For a production dashboard that may spin up 20 agents, **thread‑pool + async wrapper** offers the best of both worlds: each request runs in its own worker thread, while the underlying HTTP call stays async and streams.
Structuring Session States
We store three objects per agent:
class AgentSession(Base):
__tablename__ = "agents"
id = Column(Integer, primary_key=True)
name = Column(String, unique=True)
model = Column(String) # "claude-3-5-sonnet-202406"
api_key = Column(String) # encrypted in real prod
system_prompt = Column(String) # initial system message
last_message_id = Column(String, nullable=True)
quota_used = Column(Integer, default=0) # per‑model counters
memory_json = Column(JSON) # serialized conversation history
When an agent sends a message we:
- Load its `memory_json`.
- Append the user message.
- Call Claude with the full list.
- Stream the response back, appending each chunk to the UI and to `memory_json`.
- Persist `quota_used` based on response headers (`anthropic-ratelimit-remaining`).
Centralized Logging, Error Broker, and API Quota Manager
A singleton `DashboardCore` glues everything together:
class DashboardCore:
def __init__(self):
self.db = SessionLocal()
self.executor = ThreadPoolExecutor(max_workers=12)
self.quota_lock = threading.Lock()
self.logger = logger = loguru.logger
self.active_agents: dict[int, AgentSession] = {}
def submit_job(self, agent_id: int, fn: Callable, *args, **kwargs):
"""Wraps job submission with quota checks and exception safety."""
def wrapped():
try:
with self.quota_lock:
# quick check – allow if remaining > 0 else queue
if not self._can_send(agent_id):
raise RuntimeError("Quota exhausted, request throttled")
return fn(*args, **kwargs)
except Exception as exc:
self.logger.exception(f"Agent {agent_id} failed")
# Propagate a safe sentinel to UI
return {"error": str(exc)}
return self.executor.submit(wrapped)
def _can_send(self, agent_id: int) -> bool:
sess = self.db.get(AgentSession, agent_id)
return sess.quota_used < sess.quota_limit # limit read from model metadata
All UI threads call `DashboardCore.submit_job`; the worker returns a dict that UI can render (either a response chunk or an error object).
—
Step‑by‑Step Build: The Main Dashboard GUI
1. Setting Up the Master Event Loop and Thread Pool
Create a small bootstrap script:
# bootstrap.sh
#!/usr/bin/env bash
python - <<'PY'
# python 3.12 required
import sys, pathlib; sys.path.append(str(pathlib.Path(__file__).parent))
import main
if __name__ == "__main__":
main.run()
PY
PY
# main.py
# version: 2026.10.03
import sys
import asyncio
from pathlib import Path
from PyQt6 import QtWidgets, QtCore
from core import DashboardCore, AgentWindow
def run():
app = QtWidgets.QApplication(sys.argv)
core = DashboardCore()
main_win = AgentWindow(core)
main_win.show()
# Integrate asyncio loop for streaming
loop = asyncio.get_event_loop()
loop.run_until_complete(main_win.startup())
sys.exit(app.exec())
`AgentWindow` builds the UI; we’ll flesh it out next.
2. Building the UI: Agent List, Chat Panels, and Global Controls
# ui.py
# version: 2026.10.03
from PyQt6 import QtWidgets, QtCore
from core import DashboardCore
class AgentWindow(QtWidgets.QMainWindow):
def __init__(self, core: DashboardCore):
super().__init__()
self.core = core
self.setWindowTitle("Claude Multi‑Agent Dashboard")
self.resize(1200, 800)
self._setup_ui()
def _setup_ui(self):
central = QtWidgets.QWidget()
self.setCentralWidget(central)
layout = QtWidgets.QHBoxLayout(central)
# Left pane – agent list + controls
self.agent_list = QtWidgets.QListWidget()
self.agent_list.setMaximumWidth(250)
self.agent_list.itemClicked.connect(self._switch_agent)
layout.addWidget(self.agent_list)
# Right pane – stacked chat widgets
self.chat_stack = QtWidgets.QStackedWidget()
layout.addWidget(self.chat_stack)
# Bottom global toolbar
toolbar = QtWidgets.QToolBar()
self.addToolBar(QtCore.Qt.ToolBarArea.BottomToolBarArea, toolbar)
self.add_btn = QtWidgets.QPushButton("Add Agent")
self.add_btn.clicked.connect(self._add_agent_dialog)
toolbar.addWidget(self.add_btn)
async def startup(self):
# Load persisted agents
agents = self.core.db.query(AgentSession).all()
for a in agents:
self._register_agent(a)
def _register_agent(self, sess):
# List entry
item = QtWidgets.QListWidgetItem(sess.name)
item.setData(QtCore.Qt.ItemDataRole.UserRole, sess.id)
self.agent_list.addItem(item)
# Chat pane
chat = ChatPane(sess, self.core)
self.chat_stack.addWidget(chat)
def _switch_agent(self, item):
aid = item.data(QtCore.Qt.ItemDataRole.UserRole)
idx = self.chat_stack.indexOf(self.chat_stack.findChild(ChatPane, f"chat_{aid}"))
if idx != -1:
self.chat_stack.setCurrentIndex(idx)
def _add_agent_dialog(self):
dlg = AddAgentDialog(self.core, self)
dlg.exec()
`ChatPane` holds a `QTextEdit` for history, an `QLineEdit` for user input, and a **Cancel** button that signals the worker thread to abort.
3. Integrating the Claude API Client with Tool‑Use and File Handling
# client.py
# version: 2026.10.03
import aiohttp
import asyncio
from anthropic import Anthropic, AsyncClient
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
from pydantic import BaseModel, ValidationError
from core import DashboardCore
# ----------------------------------------------------------------------
# Retry policy shared across all agents
# ----------------------------------------------------------------------
@retry(
reraise=True,
stop=stop_after_attempt(5),
wait=wait_exponential_jitter(initial=1, max=30),
retry=retry_if_exception_type((aiohttp.ClientError, RuntimeError)),
)
async def call_claude(client: AsyncClient, payload: dict):
async with client.messages.stream(**payload) as stream:
async for chunk in stream:
yield chunk
# ----------------------------------------------------------------------
# Agent worker – runs inside ThreadPoolExecutor
# ----------------------------------------------------------------------
def agent_worker(core: DashboardCore, agent_id: int, user_msg: str):
sess = core.db.get(AgentSession, agent_id)
client = AsyncClient(
api_key=sess.api_key,
timeout=120, # generous for tool‑use
max_retries=0, # we handle retries ourselves
)
# Build full message list
messages = sess.memory_json or []
messages.append({"role": "user", "content": user_msg})
payload = {
"model": sess.model,
"messages": messages,
"max_tokens": 2048,
"stream": True,
}
# If the agent has Computer Use enabled, add tool spec
if "computer_use" in sess.system_prompt.lower():
payload["tools"] = [{"name": "computer_use"}]
async def runner():
response_chunks = []
async for chunk in call_claude(client, payload):
# chunk is a dict with "type": "content_block_delta"
delta = chunk.get("delta", {})
if "text" in delta:
response_chunks.append(delta["text"])
# Send incremental update back to UI via core logger
core.logger.info(f"[{sess.name}] {delta['text']}")
# Update persisted session
full_response = "".join(response_chunks)
messages.append({"role": "assistant", "content": full_response})
sess.memory_json = messages
# Update quota counters from response headers
remaining = client.last_response.headers.get("anthropic-ratelimit-remaining")
if remaining is not None:
sess.quota_used = int(remaining)
core.db.commit()
return {"response": full_response}
# Run the async runner in its own loop
return asyncio.run(runner())
**Tip:** The `call_claude` function streams directly from the HTTP response, avoiding the need to buffer the whole reply. This keeps latency low and lets the UI display words as they arrive.
—
Engineering Robust Parallel Agents
Managing Independent Conversation Contexts
Each `AgentSession` owns a `memory_json` list. We never share this list across threads; the worker loads a fresh copy, mutates it, and writes it back under the DB lock. This eliminates race conditions even when two agents use the same SQLite file.
# inside agent_worker
with core.db.begin():
sess = core.db.get_for_update(AgentSession, agent_id) # SELECT … FOR UPDATE
# safe mutation now
Implementing Fault Tolerance: Retry Logic, Exponential Backoff, and Timeouts
The `@retry` decorator from Tenacity handles transient network blips. For rate‑limit (429) responses we raise a custom exception to trigger a longer back‑off:
class RateLimitExceeded(RuntimeError):
pass
@retry(
retry=retry_if_exception_type(RateLimitExceeded),
wait=wait_exponential_jitter(initial=5, max=120),
stop=stop_after_attempt(3),
)
async def call_claude(...):
async with client.messages.stream(**payload) as stream:
if stream.response.status == 429:
raise RateLimitExceeded("Quota exhausted")
async for chunk in stream:
yield chunk
When a worker finally gives up, the `DashboardCore.submit_job` returns an `error` dict. The UI renders it in red and disables the offending agent until the user clicks **Reset Quota**.
Real‑Time Response Streaming to Multiple UI Widgets Without Freeze
Qt’s signal/slot system is thread‑safe when using `QtCore.pyqtSignal`. The worker emits a custom `chunk