Claude AI GUI Agent: Build Real‑Time Code Review Tool (2026)

When I first tried to glue a Claude 3.5 Sonnet agent to a desktop app, the CLI version I’d written for a quick proof‑of‑concept slammed the terminal with a massive JSON blob and then froze for ten seconds while the API throttled. The real pain‑point wasn’t Claude—it was the missing UI glue: no progress bar, no way to cancel a runaway tool call, and no place for the developer to look at the exact line that triggered a warning. The solution is to give the agent a proper graphical front‑end that streams tokens, keeps the event loop alive, and lets a human intervene before any destructive action happens.

Build a Claude Agent GUI for Static Analysis (2026)

A semi-automated code review workflow with Claude involves a desktop GUI where a developer picks files. The GUI calls the Claude API to run static analysis via tool calls, streams the agent’s findings, and lets the user approve, reject, or ask clarifications in real time, merging AI speed with human judgment.

The Semi‑Automated Code Review Architecture (2026)

The system splits into three logical layers:

1. **Agent Orchestrator** – owns the Claude session, formats prompts, interprets tool‑call responses, and retries on rate limits.
2. **State Manager** – persists the conversation (messages, file snapshots, Git hashes) in a tiny SQLite DB, so a multi‑step review never loses context.
3. **Inference Engine** – the asynchronous bridge that sends requests to Claude, handles streaming tokens, and forwards them to the UI.

flowchart LR
    UI[Desktop GUI] -->|user action| ORCH[Agent Orchestrator]
    ORCH -->|read/write| SM[State Manager]
    ORCH -->|async request| INF[Inference Engine]
    INF -->|stream tokens| UI
    INF -->|tool call| TOOL[File I/O / GitPython]
    TOOL -->|result| INF

Why GUI over CLI? The Human‑in‑the‑Loop Advantage

A CLI can dump a JSON payload, but a developer needs:

• **Live token streaming** – see reasoning as it unfolds instead of waiting for the whole response.
• **Cancellation** – a “Stop” button aborts a long‑running tool call before a mistaken file deletion.
• **Contextual navigation** – click a line number in the findings panel and jump straight to the editor view.

These interactions turn a raw LLM output into an *explainable* review process that pairs AI speed with human judgment.

Core Components

Component	Responsibility	Tech
**Agent Orchestrator**	Prompt templating, tool‑call routing, retry/back‑off, session token handling	`anthropic` SDK, `asyncio`
**State Manager**	SQLite DB (`sqlite3`), models for messages, file snapshots, git refs	`sqlite3`, `pydantic`
**Inference Engine**	Async HTTP streaming, token forwarding, error classification	`httpx` (used internally by SDK)
**GUI Layer**	PyQt6 widgets, qasync integration, progress & cancellation UI	`PyQt6`, `qasync`

Project Setup & Prerequisites

# Create a clean virtual environment
python -m venv .venv
source .venv/bin/activate

# Install the stack
pip install "anthropic==0.11.*" \
            "pyqt6==6.6.*" \
            "qasync==0.27.*" \
            "gitpython==3.1.*" \
            "pydantic==2.*"

Claude API Keys & Environment (2026 SDK)

The 2026 SDK expects the key in `ANTHROPIC_API_KEY`. For added safety, use the OS secret store (e.g., `keyring` on macOS/Linux or Windows Credential Manager).

import keyring, os

if not os.getenv("ANTHROPIC_API_KEY"):
    key = keyring.get_password("anthropic", "api_key")
    if key:
        os.environ["ANTHROPIC_API_KEY"] = key
    else:
        raise RuntimeError("Claude API key not found")

Choosing Your GUI Framework

• **PyQt6** – full‑featured desktop app, native look‑and‑feel, excellent for complex widgets (file tree, tables).
• **Tkinter** – lightweight but limited styling; not recommended for token streaming.
• **Streamlit** – great for quick prototypes, but suffers from the same blocking issue we’re solving here.

We’ll stick with **PyQt6** and `qasync` to keep the UI responsive while using `asyncio` under the hood.

Building the Claude Agent Orchestrator

Structured Prompt Design for Code Analysis

The prompt must be deterministic: give Claude a system message that defines the role, the tools, and the expected JSON schema for findings.

SYSTEM_PROMPT = """You are Claude, an expert static‑analysis assistant.
You have access to the following tools:
- `read_file(path)` – returns the raw text of a file.
- `list_repo()` – returns a list of tracked files.
- `run_git(command)` – executes a Git command and returns stdout.

When you spot a potential issue, respond with a JSON block:
{
  "file": "<relative_path>",
  "line": <int>,
  "severity": "low|medium|high",
  "message": "<human‑readable description>"
}
Always wrap JSON in triple backticks so the orchestrator can parse it."""

Implementing Tool Calling for File I/O & Git

Anthropic’s SDK will return a `tool_use` message when Claude asks for a tool. We map those to concrete Python functions.

# orchestrator.py
import asyncio, json, os, pathlib, logging
from anthropic import AsyncAnthropic, RateLimitError, APIError
from git import Repo
from pydantic import BaseModel, ValidationError

log = logging.getLogger(__name__)

# Simple data model for a finding
class Finding(BaseModel):
    file: str
    line: int
    severity: str
    message: str

class ClaudeOrchestrator:
    def __init__(self, repo_path: str):
        self.client = AsyncAnthropic()
        self.repo = Repo(repo_path)
        self.session = []                     # List[dict] of messages
        self.state_db = None                  # Filled later by StateManager

    async def _call_tool(self, name: str, args: dict):
        try:
            if name == "read_file":
                path = pathlib.Path(self.repo.working_tree_dir) / args["path"]
                return path.read_text()
            elif name == "list_repo":
                return [item.path for item in self.repo.tree().traverse() if item.type == "blob"]
            elif name == "run_git":
                result = self.repo.git.execute(args["command"].split())
                return result
            else:
                raise ValueError(f"Unknown tool: {name}")
        except Exception as exc:
            log.exception("Tool execution failed")
            return f"Error: {exc}"

    async def _stream_response(self, prompt: str, on_token):
        """Send a prompt and feed each token to on_token callback."""
        backoff = 1
        while True:
            try:
                async with self.client.messages.stream(
                    model="claude-3-5-sonnet-202406",
                    max_tokens=4096,
                    temperature=0.0,
                    system=SYSTEM_PROMPT,
                    messages=self.session + [{"role": "user", "content": prompt}],
                ) as stream:
                    async for event in stream:
                        if event.type == "content_block_start":
                            continue
                        if event.type == "content_block_delta":
                            await on_token(event.delta.get("text", ""))
                        if event.type == "tool_use":
                            tool_name = event.tool_use.name
                            tool_args = json.loads(event.tool_use.input)
                            result = await self._call_tool(tool_name, tool_args)
                            # Feed tool result back into the conversation
                            self.session.append({
                                "role": "assistant",
                                "content": [{"type": "tool_result", "tool_use_id": event.tool_use.id,
                                            "content": result}]
                            })
                    # Normal exit, break retry loop
                    break
            except RateLimitError as e:
                log.warning("Rate limit hit, backing off %ss", backoff)
                await asyncio.sleep(backoff)
                backoff = min(backoff * 2, 60)
            except APIError as e:
                log.error("API error: %s", e)
                raise

State Management: Tracking Conversations & Analysis Context

A tiny SQLite file lives beside the project folder. The `StateManager` encapsulates CRUD for messages and findings, ensuring we can resume a review after a crash.

# state_manager.py
import sqlite3, json, datetime
from typing import List
from pydantic import BaseModel

class DB:
    def __init__(self, db_path: str):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self._init_schema()

    def _init_schema(self):
        cur = self.conn.cursor()
        cur.executescript("""
        CREATE TABLE IF NOT EXISTS messages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            role TEXT NOT NULL,
            content TEXT NOT NULL,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
        );
        CREATE TABLE IF NOT EXISTS findings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            file TEXT NOT NULL,
            line INTEGER NOT NULL,
            severity TEXT NOT NULL,
            message TEXT NOT NULL,
            created_at DATETIME DEFAULT CURRENT_TIMESTAMP
        );
        """)
        self.conn.commit()

    def add_message(self, role: str, content: str):
        self.conn.execute(
            "INSERT INTO messages (role, content) VALUES (?, ?)",
            (role, content),
        )
        self.conn.commit()

    def get_messages(self) -> List[dict]:
        cur = self.conn.execute("SELECT role, content FROM messages ORDER BY id")
        return [{"role": r, "content": c} for r, c in cur.fetchall()]

    def add_finding(self, finding: dict):
        self.conn.execute(
            "INSERT INTO findings (file, line, severity, message) VALUES (:file, :line, :severity, :message)",
            finding,
        )
        self.conn.commit()

    def list_findings(self) -> List[dict]:
        cur = self.conn.execute("SELECT file, line, severity, message FROM findings ORDER BY severity DESC")
        return [dict(zip([c[0] for c in cur.description], row)) for row in cur.fetchall()]

The orchestrator loads the persisted `session` on startup:

# inside ClaudeOrchestrator.__init__
self.state_db = DB(os.path.join(repo_path, ".claude_review.db"))
self.session = self.state_db.get_messages()

Whenever we receive a tool result or a final finding JSON block, we write it back:

# after parsing a JSON finding
try:
    f = Finding.parse_raw(json_block)
    self.state_db.add_finding(f.dict())
except ValidationError as ve:
    log.warning("Invalid finding format: %s", ve)

Handling Rate Limits & API Timeouts Gracefully

The `_stream_response` method already implements exponential back‑off for `RateLimitError`. For timeouts we add a per‑call `asyncio.wait_for`.

async def safe_stream(self, prompt, on_token, timeout=120):
    try:
        await asyncio.wait_for(self._stream_response(prompt, on_token), timeout)
    except asyncio.TimeoutError:
        log.error("Claude timed out after %s seconds", timeout)
        # Notify UI via a dedicated signal (see GUI section)

Designing the Interactive Desktop GUI

Project Selection & File Tree Widget

We let the user browse a local repo with `QFileSystemModel` limited to the repo root. Double‑clicking a file loads its content into the “source” pane.

# gui.py
import os, pathlib, sys, asyncio
from PyQt6.QtWidgets import (
    QApplication, QMainWindow, QTreeView, QTextEdit,
    QPushButton, QSplitter, QWidget, QVBoxLayout, QLabel,
    QProgressBar, QMessageBox,
)
from PyQt6.QtCore import Qt, QItemSelectionModel, pyqtSignal, QObject
from qasync import QEventLoop, asyncSlot

class Communicator(QObject):
    token_received = pyqtSignal(str)
    analysis_done = pyqtSignal()
    error_occurred = pyqtSignal(str)

class MainWindow(QMainWindow):
    def __init__(self, repo_path: str):
        super().__init__()
        self.setWindowTitle("Claude Code Review")
        self.resize(1200, 800)

        self.repo_path = pathlib.Path(repo_path).resolve()
        self.comm = Communicator()
        self._build_ui()
        self._connect_signals()
        self.orchestrator = None  # Filled after async init

    def _build_ui(self):
        splitter = QSplitter(Qt.Orientation.Horizontal)

        # Left: File tree
        self.tree = QTreeView()
        model = QFileSystemModel()
        model.setRootPath(str(self.repo_path))
        model.setFilter(Qt.MatchFlag.MatchRecursive | Qt.MatchFlag.MatchCaseSensitive)
        self.tree.setModel(model)
        self.tree.setRootIndex(model.index(str(self.repo_path)))
        self.tree.hideColumn(1)  # size
        self.tree.hideColumn(2)  # type
        self