When you hand an autonomous Claude 3.5 Sonnet the job of “run the tests, build the image, push to the registry and deploy”, a simple one‑liner Bash script quickly turns into a brittle nightmare. A forgotten `cd` or a stray `SIGINT` leaves your repo in a half‑merged state, Docker hangs, and you spend more time untangling the mess than actually delivering value.
That’s why I built a **Claude CI/CD Agent GUI** – a desktop app that reads natural‑language intent, orchestrates Git, Docker, and arbitrary shell commands, and streams live logs back to a responsive PyQt6 interface. The result is a single click that replaces a dozen manual steps, with full visibility and a safety net you can abort at any moment.
- Claude Desktop + Python 3.11+ powers a stateful, async CI/CD assistant.
- PyQt6’s QThread + signal/slot pattern streams logs without freezing the UI.
- Tool registry keeps execution context (working directory, build ID) across calls.
- Built‑in retry, back‑off, and a “kill switch” make the agent production‑ready.
- Package with PyInstaller or Tauri for a distributable binary.
Before you start: Install Python 3.11+, `pip install anthropic[async] pyqt6 gitpython docker python-dotenv`. Grab a Claude API key (3.5 Sonnet or Haiku) and enable **Computer Use** in Claude Desktop. Create a virtual environment and copy a `.env` template with `ANTHROPIC_API_KEY` and `DOCKER_HOST`.
What is a Claude CI/CD Agent?
A Claude CI/CD agent is a desktop GUI application that uses the Claude API, Python’s `asyncio`, and a framework like PyQt to automate local development scripts. It listens to natural language commands like “deploy feature X,” then autonomously executes the necessary Git, Docker, and shell operations, streaming progress and handling errors interactively.
System Architecture & Workflow
flowchart LR
UI[GUI (PyQt6)] -->|user input| Agent[Async Agent Core]
Agent -->|tool request| Registry[Tool Registry]
Registry --> Git[GitPython]
Registry --> Docker[Docker Engine API]
Registry --> Shell[Shell Exec]
Git -->|status| UI
Docker -->|logs| UI
Shell -->|output| UI
Agent -->|events| UI
UI -->|kill| Agent
- **UI (PyQt6)** – presents a chat box, log view, and buttons.
- **Async Agent Core** – runs an `asyncio` event loop, calls Claude, parses tool calls.
- **Tool Registry** – maps Claude‑requested tool names to Python wrappers that preserve a mutable `AgentState` (working dir, build IDs).
- **Git/Docker/Shell** – concrete implementations using `GitPython`, Docker Engine HTTP API, and `asyncio.create_subprocess_exec`.
- **Bidirectional signaling** – `QThread` emits `log_line` signals back to the main thread; the UI updates instantly without blocking.
Architecting the Reactive Agent Core
Async Event Loop & Tool Registry
# agent_core.py – v1.0
import asyncio
import os
from anthropic import AsyncAnthropic, HUMAN_PROMPT, AI_PROMPT
from typing import Dict, Callable, Any
from dataclasses import dataclass, field
API_KEY = os.getenv("ANTHROPIC_API_KEY")
client = AsyncAnthropic(api_key=API_KEY)
@dataclass
class AgentState:
cwd: str = os.getcwd()
build_id: str | None = None
env: dict = field(default_factory=lambda: os.environ.copy())
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, Callable[[AgentState, Dict[str, Any]], asyncio.Future]] = {}
def register(self, name: str):
def decorator(fn):
self._tools[name] = fn
return fn
return decorator
async def call(self, name: str, state: AgentState, args: dict):
if name not in self._tools:
raise ValueError(f"Tool {name} is not registered")
return await self._tools[name](state, args)
registry = ToolRegistry()
**Why a mutable `AgentState`?** Claude can ask “run the Docker build in the directory we just checked out”. Keeping `cwd` and `build_id` inside a shared object lets sequential tool calls share context without re‑parsing the entire prompt.
Defining Critical CI/CD Tools
# git_tool.py – v1.0
import git
from agent_core import registry, AgentState
@registry.register("git")
async def git_tool(state: AgentState, args: dict):
repo = git.Repo(state.cwd)
cmd = args.get("command")
if cmd == "checkout":
branch = args["branch"]
repo.git.checkout(branch)
return f"Checked out {branch}"
if cmd == "status":
return repo.git.status()
raise ValueError(f"Unsupported Git command: {cmd}")
# docker_tool.py – v1.0
import aiohttp
import json
from agent_core import registry, AgentState
DOCKER_API = os.getenv("DOCKER_HOST", "unix://var/run/docker.sock")
@registry.register("docker")
async def docker_tool(state: AgentState, args: dict):
async with aiohttp.UnixConnector(path="/var/run/docker.sock") if DOCKER_API.startswith("unix://") else aiohttp.TCPConnector() as conn:
async with aiohttp.ClientSession(connector=conn) as session:
if args["action"] == "build":
tag = args["tag"]
dockerfile = args.get("dockerfile", "Dockerfile")
build_context = state.cwd
# Stream build output line‑by‑line
async with session.post(
f"{DOCKER_API}/build",
params={"t": tag, "dockerfile": dockerfile},
data=open(os.path.join(build_context, "tarball.tar"), "rb"),
headers={"Content-Type": "application/x-tar"}
) as resp:
async for line in resp.content:
decoded = line.decode()
# Pass raw JSON line to UI via state logger
state.env["log_callback"](decoded)
return f"Built image {tag}"
raise ValueError("Unsupported Docker action")
**My take:** I wired Docker through the raw Engine API instead of the `docker` CLI because it surfaces structured JSON logs that parse cleanly into the UI. The performance gain is negligible, but the observability boost is huge.
State Management for Long‑Running Operations
# utils.py – v1.0
import asyncio
from typing import Callable
class LogForwarder:
"""Thread‑safe bridge between async tool code and PyQt signals."""
def __init__(self, loop: asyncio.AbstractEventLoop, callback: Callable[[str], None]):
self.loop = loop
self.callback = callback
def emit(self, line: str):
if asyncio.iscoroutinefunction(self.callback):
asyncio.run_coroutine_threadsafe(self.callback(line), self.loop)
else:
self.loop.call_soon_threadsafe(self.callback, line)
When initializing the agent we inject `log_forwarder.emit` into `AgentState.env[“log_callback”]`. Every tool can now push incremental logs without caring about the GUI threading model.
Crafting the Interactive Desktop GUI (PyQt Focus)
Layout: Log View, Tool Panel, and Chat Interface
# gui.py – v1.0
import sys, os
from PyQt6.QtWidgets import (QApplication, QWidget, QVBoxLayout, QTextEdit,
QLineEdit, QPushButton, QHBoxLayout)
from PyQt6.QtCore import QThread, pyqtSignal, QObject
from agent_core import AgentState, registry, client
from utils import LogForwarder
import asyncio
class Worker(QObject):
log_line = pyqtSignal(str)
finished = pyqtSignal()
def __init__(self, prompt: str, state: AgentState):
super().__init__()
self.prompt = prompt
self.state = state
async def run_async(self):
async def callback(line: str):
self.log_line.emit(line)
# attach forwarder
self.state.env["log_callback"] = callback
# Simple loop: send prompt, get tool calls, execute, repeat until Claude says "DONE"
messages = [{"role": "user", "content": self.prompt}]
while True:
response = await client.messages.create(
model="claude-3-5-sonnet-202406",
max_tokens=1024,
messages=messages,
tools=registry._tools.keys(),
temperature=0.1,
)
if response.stop_reason == "tool_use":
tool_call = response.content[0].input
tool_name = tool_call.name
args = tool_call.input
try:
result = await registry.call(tool_name, self.state, args)
self.log_line.emit(f"[{tool_name}] {result}")
except Exception as e:
self.log_line.emit(f"[ERROR] {tool_name}: {e}")
messages.append(response)
else:
# Claude finished reasoning
self.log_line.emit("[AI] " + response.content[0].text)
break
self.finished.emit()
def start(self):
asyncio.run(self.run_async())
class MainWindow(QWidget):
def __init__(self):
super().__init__()
self.setWindowTitle("Claude CI/CD Agent")
self.resize(800, 600)
self.layout = QVBoxLayout(self)
self.log_view = QTextEdit(self)
self.log_view.setReadOnly(True)
self.input_line = QLineEdit(self)
self.send_btn = QPushButton("Send", self)
self.send_btn.clicked.connect(self.handle_send)
bottom = QHBoxLayout()
bottom.addWidget(self.input_line)
bottom.addWidget(self.send_btn)
self.layout.addWidget(self.log_view)
self.layout.addLayout(bottom)
# Agent state lives across invocations
self.agent_state = AgentState()
def handle_send(self):
prompt = self.input_line.text().strip()
if not prompt:
return
self.log_view.append(f"> {prompt}")
self.input_line.clear()
self.run_agent(prompt)
def run_agent(self, prompt: str):
self.thread = QThread()
self.worker = Worker(prompt, self.agent_state)
self.worker.moveToThread(self.thread)
self.worker.log_line.connect(self.append_log)
self.worker.finished.connect(self.thread.quit)
self.thread.started.connect(self.worker.start)
self.thread.start()
def append_log(self, line: str):
self.log_view.append(line)
if __name__ == "__main__":
app = QApplication(sys.argv)
win = MainWindow()
win.show()
sys.exit(app.exec())
**Key points**
- **QThread + pyqtSignal** guarantees the UI stays responsive while the async agent runs.
- `LogForwarder` routes each incremental line from the tool wrappers straight to `log_line`, which the main thread appends to the QTextEdit.
- The `AgentState` object persists across multiple user commands, so a “checkout feature branch” followed by “docker build” reuses the same `cwd`.
Tip: Set `self.thread.setTerminationEnabled(True)` if you need to abort a runaway Docker build from the UI – just call `self.thread.terminate()` inside a “Stop” button handler.
Interactive Controls: Triggering and Monitoring Deployments
Add a **Deploy** button that pre‑populates a prompt template:
self.deploy_btn = QPushButton("Deploy Feature", self)
self.deploy_btn.clicked.connect(lambda: self.handle_send(
"Deploy the current branch to staging. Use git checkout, docker build with tag staging-<commit>, and push to registry."))
self.layout.addWidget(self.deploy_btn)
Now a single click spawns the full Claude‑driven chain, while the log view streams each Git status line, Docker build step, and final confirmation.
Production‑Proofing: Error Handling & Observability
Graceful API Failure, Rate Limit, and Timeout Recovery
# retry_utils.py – v1.0
import asyncio
from typing import Callable, Awaitable
import random
async def exponential_backoff(fn: Callable[[], Awaitable], retries: int = 5):
delay = 1
for attempt in range(retries):
try:
return await fn()
except Exception as exc:
if "429" in str(exc) or "rate limit" in str(exc).lower():
jitter = random.uniform(0, 0.5)
await asyncio.sleep(delay + jitter)
delay = min(delay * 2, 30)
else:
raise
raise RuntimeError("Maximum retries exceeded")
Wrap every Claude request:
response = await exponential_backoff(
lambda: client.messages.create(
model="claude-3-5-sonnet-202406",
max_tokens=1024,
messages=messages,
tools=registry._tools.keys(),
temperature=0.1,
)
)
Logging, Retry Logic, and Operation Rollbacks
- Git failures → `git reset –hard ORIG_HEAD`.
- Docker build failure → `docker image rm
`.
@registry.register("git")
async def git_tool(state: AgentState, args: dict):
repo = git.Repo(state.cwd)
try:
# existing logic...
return "Success"
except git.exc.GitCommandError as e:
repo.git.reset('--hard', 'ORIG_HEAD')
raise RuntimeError(f"Git error, rolled back: {e}")
Adding a “Kill Switch” and State Persistence
# persistence.py – v1.0
import json, pathlib
SAVE_PATH = pathlib.Path.home() / ".claude_ci_state.json"
def save_state(state: AgentState):
with open(SAVE_PATH, "w") as f:
json.dump({"cwd": state.cwd, "build_id": state.build_id}, f)
def load_state() -> AgentState:
if SAVE_PATH.exists():
data = json.load(open(SAVE_PATH))
return AgentState(**data)
return AgentState()
Load at startup:
self.agent_state = load_state()
Add a **Kill** button that sets a flag in `AgentState.env[“cancel”] = True`; each tool checks this flag before proceeding and raises `asyncio.CancelledError` if set.
Warning: Never invoke `os._exit` from a worker thread. Use the cancellation flag and let the async loop unwind gracefully, otherwise you’ll corrupt the Docker daemon socket.
From Prototype to Daily Driver: Advanced Patterns
Multi‑Step Prompt Chaining for Complex Deployments
Claude can emit a series of tool calls that depend on prior results. By preserving `AgentState` you enable patterns like:
- `git checkout` → updates `cwd`.
- `docker build` reads `state.cwd` and tags with the commit hash stored in `state.build_id`.
- `shell exec` runs `kubectl rollout restart` with the new image tag.
Integrating with Webhook Notifications (Slack, Teams)
@registry.register("slack_notify")
async def slack_notify(state: AgentState, args: dict):
import httpx
webhook = os.getenv("SLACK_WEBHOOK")
await httpx.AsyncClient().post(webhook, json={"text": args["message"]})
return "Notified Slack"
You can now ask Claude: “After the deployment succeed, ping #devops.” The agent will automatically fire the webhook.
Security Hardening: Input Sanitization and Secret Management
- Store secrets in OS‑level keyring (`keyring` Python lib) instead of `.env`.
- Whitelist allowed shell commands:
ALLOWED_CMDS = {"kubectl", "helm", "aws"}
if args["cmd"] not in ALLOWED_CMDS:
raise PermissionError(f"Command {args['cmd']} is not permitted")
- Run the whole agent under a dedicated Unix user (`claude-agent`) with `Docker` group membership only.
For a deeper dive on secret handling, see my post on **[Local LLM (Llama) vs. Claude API: Hybrid Agents for Cost Control (2026)](https://nileshblog.tech/?p=7209)**.
Packaging Your App: PyInstaller or Native Builds
# PyInstaller one‑file bundle
pip install pyinstaller
pyinstaller --onefile --windowed gui.py -n ClaudeCICDAgent
If you prefer a lighter web‑based shell, Tauri v2 can embed the same Python backend via a Rust bridge; the UI is built with HTML/CSS/React. The core logic remains unchanged because the agent