Seth/workbench-server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
28dc693781ad143451d1f05a472ed3405d871df3
workbench-server/docs/superpowers/plans/2026-03-30-workbench-v2.md
T
Mortdecai 28dc693781 docs: workbench v2 implementation plan — 8 tasks
2026-03-30 07:19:20 -04:00

1226 lines
38 KiB
Markdown
Raw Blame History

# Workbench Server v2 Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Restructure workbench-server into a pip-installable package with server persistence (survives AI CLI restarts), simplified desktop-only layout, and AI-guided setup.
**Architecture:** The existing monolithic `server.py` gets split into a proper Python package under `src/workbench/`. A `.server.json` file per project tracks running HTTP servers. On MCP startup, the server scans for live servers and reattaches. The scaffold HTML drops the terminal panel for a full-width display surface.
**Tech Stack:** Python 3.10+, `mcp` (FastMCP), `aiohttp` (HTTP/WebSocket), `pytest`/`pytest-asyncio` for tests.
**Spec:** `docs/superpowers/specs/2026-03-30-workbench-v2-design.md`
**Existing code:** The current `server.py` (293 lines) and `scaffold.html` (142 lines) are the starting point. We're restructuring, not rewriting from scratch — the core logic (MCP tools, HTTP server, WebSocket broadcast) is preserved.
---
## File Structure
```
workbench-server/
pyproject.toml # Package config, entry point
README.md # Public-facing docs (rewritten)
INSTALL.md # AI-readable setup instructions
LICENSE # MIT
src/
workbench/
__init__.py # Version string
__main__.py # python -m workbench
cli.py # CLI: mcp, serve, list, help
server.py # MCP tools + HTTP/WS server management
project.py # Project dir management, logging, persistence
scaffold.html # HTML template (moved from root, simplified)
tests/
conftest.py # Shared fixtures
test_project.py # Project creation, logging, persistence
test_server.py # MCP tool tests
# Old files to remove:
server.py # replaced by src/workbench/server.py
scaffold.html # replaced by src/workbench/scaffold.html
requirements.txt # replaced by pyproject.toml
```
---
## Task 1: Package Scaffold + Move Files
**Files:**
- Create: `pyproject.toml`
- Create: `LICENSE`
- Create: `src/workbench/__init__.py`
- Create: `src/workbench/__main__.py`
- Remove: `requirements.txt`
- Move: `server.py` → will be replaced in later tasks
- Move: `scaffold.html` → will be replaced in later tasks
- [ ] **Step 1: Create directory structure**
```bash
mkdir -p ~/bin/workbench-server/src/workbench
mkdir -p ~/bin/workbench-server/tests
```
- [ ] **Step 2: Write pyproject.toml**
```toml
[build-system]
requires = ["setuptools>=68.0"]
build-backend = "setuptools.build_meta"
[project]
name = "workbench-server"
version = "0.2.0"
description = "MCP server that lets AI CLIs build interactive web pages served over LAN"
readme = "README.md"
license = "MIT"
requires-python = ">=3.10"
dependencies = [
"mcp>=1.26.0",
"aiohttp>=3.9.0",
]
[project.scripts]
workbench = "workbench.cli:main"
```
- [ ] **Step 3: Write __init__.py**
```python
"""Workbench — MCP server for AI-driven interactive web pages."""
__version__ = "0.2.0"
```
- [ ] **Step 4: Write __main__.py**
```python
"""Allow running as `python -m workbench`."""
from workbench.cli import main
main()
```
- [ ] **Step 5: Write LICENSE (MIT)**
- [ ] **Step 6: Back up old files, remove requirements.txt**
```bash
mkdir -p ~/bin/workbench-server/.backup
cp ~/bin/workbench-server/server.py ~/bin/workbench-server/.backup/server.py.v1
cp ~/bin/workbench-server/scaffold.html ~/bin/workbench-server/.backup/scaffold.html.v1
rm ~/bin/workbench-server/requirements.txt
```
- [ ] **Step 7: Commit**
```bash
cd ~/bin/workbench-server
git add pyproject.toml LICENSE src/ .backup/
git rm requirements.txt
git commit -m "feat: package scaffold — pyproject.toml, src/workbench/ layout"
```
---
## Task 2: Project Management Module
**Files:**
- Create: `src/workbench/project.py`
- Create: `tests/conftest.py`
- Create: `tests/test_project.py`
Handles project directory creation, session logging, log reading, listing, and the new `.server.json` persistence.
- [ ] **Step 1: Write test file**
```python
# tests/test_project.py
import json
from workbench.project import (
create_project, project_exists, project_path,
append_log, read_log, list_projects, log_session_event,
write_server_info, read_server_info, clear_server_info,
WORKBENCH_DIR,
)
def test_create_project(tmp_workbench):
path = create_project("io102", "Heathkit IO-102", workbench_dir=tmp_workbench)
assert path.exists()
assert (path / "index.html").exists()
assert (path / "session.md").exists()
assert (path / "session.jsonl").exists()
assert (path / "cost-log.jsonl").exists()
assert (path / "state.json").exists()
assert (path / "assets").is_dir()
md = (path / "session.md").read_text()
assert "Heathkit IO-102" in md
def test_create_project_idempotent(tmp_workbench):
create_project("io102", "First", workbench_dir=tmp_workbench)
append_log("io102", "test entry", workbench_dir=tmp_workbench)
create_project("io102", "Second", workbench_dir=tmp_workbench)
entries = read_log("io102", workbench_dir=tmp_workbench)
assert len(entries) == 1
def test_project_exists(tmp_workbench):
assert not project_exists("io102", workbench_dir=tmp_workbench)
create_project("io102", "Test", workbench_dir=tmp_workbench)
assert project_exists("io102", workbench_dir=tmp_workbench)
def test_append_and_read_log(tmp_workbench):
create_project("io102", "Test", workbench_dir=tmp_workbench)
append_log("io102", "R412 measured 1.05M", data={"ohms": 1050000}, workbench_dir=tmp_workbench)
append_log("io102", "R413 OK", workbench_dir=tmp_workbench)
entries = read_log("io102", tail=10, workbench_dir=tmp_workbench)
assert len(entries) == 2
assert entries[0]["entry"] == "R412 measured 1.05M"
assert entries[0]["ohms"] == 1050000
def test_read_log_tail(tmp_workbench):
create_project("io102", "Test", workbench_dir=tmp_workbench)
for i in range(30):
append_log("io102", f"Entry {i}", workbench_dir=tmp_workbench)
entries = read_log("io102", tail=5, workbench_dir=tmp_workbench)
assert len(entries) == 5
assert entries[0]["entry"] == "Entry 25"
def test_list_projects(tmp_workbench):
assert list_projects(workbench_dir=tmp_workbench) == []
create_project("io102", "First", workbench_dir=tmp_workbench)
create_project("psu-rebuild", "Second", workbench_dir=tmp_workbench)
projects = list_projects(workbench_dir=tmp_workbench)
assert [p["name"] for p in projects] == ["io102", "psu-rebuild"]
def test_log_session_event(tmp_workbench):
create_project("io102", "Test", workbench_dir=tmp_workbench)
log_session_event("io102", "session_start", workbench_dir=tmp_workbench)
cost_log = (tmp_workbench / "io102" / "cost-log.jsonl").read_text().strip()
entry = json.loads(cost_log.split("\n")[-1])
assert entry["event"] == "session_start"
def test_server_info_write_read_clear(tmp_workbench):
create_project("io102", "Test", workbench_dir=tmp_workbench)
write_server_info("io102", pid=12345, port=8070, workbench_dir=tmp_workbench)
info = read_server_info("io102", workbench_dir=tmp_workbench)
assert info is not None
assert info["pid"] == 12345
assert info["port"] == 8070
assert "started" in info
clear_server_info("io102", workbench_dir=tmp_workbench)
assert read_server_info("io102", workbench_dir=tmp_workbench) is None
```
- [ ] **Step 2: Write conftest.py**
```python
# tests/conftest.py
import pytest
@pytest.fixture
def tmp_workbench(tmp_path):
"""Provide a temporary workbench directory."""
wb = tmp_path / "workbench"
wb.mkdir()
return wb
```
- [ ] **Step 3: Run tests to verify they fail**
```bash
cd ~/bin/workbench-server
pip install --break-system-packages -e .
pip install --break-system-packages pytest pytest-asyncio
pytest tests/test_project.py -v
```
Expected: ImportError.
- [ ] **Step 4: Write project.py**
```python
"""Project directory management — create, log, read, list, server persistence."""
from __future__ import annotations
import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
WORKBENCH_DIR = Path.home() / "workbench"
SCAFFOLD_HTML = Path(__file__).parent / "scaffold.html"
def _now_iso() -> str:
return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
def project_path(name: str, workbench_dir: Path = WORKBENCH_DIR) -> Path:
return workbench_dir / name
def project_exists(name: str, workbench_dir: Path = WORKBENCH_DIR) -> bool:
return project_path(name, workbench_dir).is_dir()
def create_project(
name: str,
title: str,
description: str = "",
workbench_dir: Path = WORKBENCH_DIR,
) -> Path:
"""Create a project directory with scaffold HTML and log files. Idempotent."""
pdir = project_path(name, workbench_dir)
pdir.mkdir(parents=True, exist_ok=True)
(pdir / "assets").mkdir(exist_ok=True)
# Write scaffold HTML (always overwrite — it's a template, not user content)
if SCAFFOLD_HTML.exists():
template = SCAFFOLD_HTML.read_text()
html = template.replace("{{TITLE}}", title)
html = html.replace("{{DESCRIPTION}}", description)
(pdir / "index.html").write_text(html)
# Only write log files if they don't exist (preserve existing session data)
md_path = pdir / "session.md"
if not md_path.exists():
md_path.write_text(f"# {title} — Session Log\n\nStarted: {_now_iso()}\n\n")
jsonl_path = pdir / "session.jsonl"
if not jsonl_path.exists():
jsonl_path.write_text("")
cost_path = pdir / "cost-log.jsonl"
if not cost_path.exists():
cost_path.write_text("")
state_path = pdir / "state.json"
if not state_path.exists():
state_path.write_text("{}")
return pdir
def append_log(
name: str,
entry: str,
data: Optional[dict] = None,
workbench_dir: Path = WORKBENCH_DIR,
) -> dict:
"""Append to session.md and session.jsonl."""
pdir = project_path(name, workbench_dir)
ts = _now_iso()
with open(pdir / "session.md", "a") as f:
f.write(f"\n### {ts}\n{entry}\n")
obj = {"ts": ts, "entry": entry}
if data:
obj.update(data)
with open(pdir / "session.jsonl", "a") as f:
f.write(json.dumps(obj) + "\n")
return obj
def read_log(
name: str, tail: int = 20, workbench_dir: Path = WORKBENCH_DIR
) -> list[dict]:
"""Read recent log entries from session.jsonl."""
jsonl_path = project_path(name, workbench_dir) / "session.jsonl"
if not jsonl_path.exists():
return []
lines = jsonl_path.read_text().strip().split("\n")
lines = [l for l in lines if l.strip()]
recent = lines[-tail:] if len(lines) > tail else lines
entries = []
for line in recent:
try:
entries.append(json.loads(line))
except json.JSONDecodeError:
entries.append({"raw": line})
return entries
def list_projects(workbench_dir: Path = WORKBENCH_DIR) -> list[dict]:
"""List all project directories."""
if not workbench_dir.exists():
return []
projects = []
for d in sorted(workbench_dir.iterdir()):
if d.is_dir() and not d.name.startswith("."):
projects.append({"name": d.name})
return projects
def log_session_event(
name: str, event: str, workbench_dir: Path = WORKBENCH_DIR, **extra
) -> None:
"""Append to cost-log.jsonl."""
pdir = project_path(name, workbench_dir)
obj = {"ts": _now_iso(), "event": event, "project": name}
obj.update(extra)
with open(pdir / "cost-log.jsonl", "a") as f:
f.write(json.dumps(obj) + "\n")
# --- Server persistence ---
def write_server_info(
name: str, pid: int, port: int, workbench_dir: Path = WORKBENCH_DIR
) -> None:
"""Write .server.json with running server info."""
pdir = project_path(name, workbench_dir)
info = {"pid": pid, "port": port, "started": _now_iso()}
(pdir / ".server.json").write_text(json.dumps(info))
def read_server_info(
name: str, workbench_dir: Path = WORKBENCH_DIR
) -> Optional[dict]:
"""Read .server.json. Returns None if not found."""
server_file = project_path(name, workbench_dir) / ".server.json"
if not server_file.exists():
return None
try:
return json.loads(server_file.read_text())
except (json.JSONDecodeError, OSError):
return None
def clear_server_info(
name: str, workbench_dir: Path = WORKBENCH_DIR
) -> None:
"""Delete .server.json."""
server_file = project_path(name, workbench_dir) / ".server.json"
if server_file.exists():
server_file.unlink()
```
- [ ] **Step 5: Copy scaffold.html to src/workbench/**
The scaffold HTML needs to exist at `src/workbench/scaffold.html` for `create_project` to find it. We'll write the v2 version in Task 3. For now, create a minimal placeholder so tests pass:
```html
<!DOCTYPE html>
<html><head><title>{{TITLE}}</title></head>
<body><h1>{{TITLE}}</h1><p>{{DESCRIPTION}}</p></body></html>
```
- [ ] **Step 6: Run tests to verify they pass**
```bash
pytest tests/test_project.py -v
```
Expected: All 9 tests pass.
- [ ] **Step 7: Commit**
```bash
git add src/workbench/project.py src/workbench/scaffold.html tests/conftest.py tests/test_project.py
git commit -m "feat: project management module — create, log, read, list, server persistence"
```
---
## Task 3: Scaffold HTML v2
**Files:**
- Create: `src/workbench/scaffold.html` (overwrite placeholder from Task 2)
- [ ] **Step 1: Write the v2 scaffold HTML**
Full-width desktop layout. No split pane, no iframe, no terminal embed. Same WebSocket reconnect and state handling as v1.
```html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{{TITLE}}</title>
<style>
:root {
--bg: #0a0f0c;
--panel-bg: #111a15;
--border: #2a3a2e;
--text: #aaccaa;
--text-dim: #557755;
--accent: #33ff66;
}
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
background: var(--bg);
color: var(--text);
font-family: monospace;
font-size: 14px;
height: 100vh;
display: flex;
flex-direction: column;
}
#content {
flex: 1;
overflow-y: auto;
padding: 16px;
}
#log-feed {
border-top: 1px solid var(--border);
padding: 8px 12px;
max-height: 200px;
overflow-y: auto;
}
#log-feed h3 {
color: var(--text-dim);
font-size: 11px;
letter-spacing: 2px;
text-transform: uppercase;
margin-bottom: 4px;
}
.log-entry {
font-size: 12px;
color: var(--text-dim);
padding: 2px 0;
}
.log-entry .log-time {
color: var(--accent);
margin-right: 8px;
}
#status {
background: var(--panel-bg);
border-top: 1px solid var(--border);
padding: 4px 12px;
font-size: 11px;
color: var(--text-dim);
}
#status.connected { color: var(--accent); }
</style>
</head>
<body>
<div id="content">
<h1 style="color: var(--accent); font-size: 18px;">{{TITLE}}</h1>
<p style="color: var(--text-dim); margin-top: 8px;">{{DESCRIPTION}}</p>
<p style="color: var(--text-dim); margin-top: 16px;">Waiting for content...</p>
</div>
<div id="log-feed">
<h3>Session Log</h3>
<div id="log-entries"></div>
</div>
<div id="status">Connecting...</div>
<script>
const WS_URL = `ws://${location.host}/ws`;
let ws;
function connect() {
ws = new WebSocket(WS_URL);
ws.onopen = () => {
document.getElementById('status').textContent = 'Connected';
document.getElementById('status').className = 'connected';
};
ws.onclose = () => {
document.getElementById('status').textContent = 'Disconnected — reconnecting...';
document.getElementById('status').className = '';
setTimeout(connect, 2000);
};
ws.onmessage = (e) => {
const msg = JSON.parse(e.data);
if (msg.type === 'state') handleState(msg.state);
if (msg.type === 'log') handleLog(msg.entry);
};
}
function handleState(state) {
if (state.template) {
document.getElementById('content').innerHTML = state.template;
}
if (state.styles) {
let el = document.getElementById('dynamic-styles');
if (!el) { el = document.createElement('style'); el.id = 'dynamic-styles'; document.head.appendChild(el); }
el.textContent = state.styles;
}
if (state.script) {
try { new Function(state.script)(); } catch(e) { console.error('Script error:', e); }
}
window.__workbench_state = state;
try { localStorage.setItem('workbench-state', JSON.stringify(state)); } catch(e) {}
}
function handleLog(entry) {
const div = document.createElement('div');
div.className = 'log-entry';
const time = new Date().toLocaleTimeString();
div.innerHTML = `<span class="log-time">${time}</span>${entry}`;
const feed = document.getElementById('log-entries');
feed.appendChild(div);
feed.scrollTop = feed.scrollHeight;
}
try {
const saved = JSON.parse(localStorage.getItem('workbench-state'));
if (saved) handleState(saved);
} catch(e) {}
connect();
</script>
</body>
</html>
```
- [ ] **Step 2: Commit**
```bash
git add src/workbench/scaffold.html
git commit -m "feat: scaffold HTML v2 — full-width desktop layout, no terminal panel"
```
---
## Task 4: MCP Server with Persistence
**Files:**
- Create: `src/workbench/server.py`
- Create: `tests/test_server.py`
The core MCP server — same 6 tools as v1, but with persistence logic. Adapted from the original `server.py` with the `.server.json` check-and-reattach flow.
- [ ] **Step 1: Write test file**
```python
# tests/test_server.py
import asyncio
import json
import os
from pathlib import Path
from unittest.mock import patch, AsyncMock, MagicMock
import pytest
from workbench.server import WorkbenchServer
@pytest.fixture
def server(tmp_workbench):
return WorkbenchServer(workbench_dir=tmp_workbench)
@pytest.mark.asyncio
async def test_scaffold_creates_project(server, tmp_workbench):
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
result = await server.workbench_scaffold("test-proj", "Test Project")
data = json.loads(result)
assert "url" in data
assert (tmp_workbench / "test-proj" / "index.html").exists()
@pytest.mark.asyncio
async def test_scaffold_reattaches_to_running_server(server, tmp_workbench):
"""If .server.json exists and server is alive, don't start a new one."""
from workbench.project import create_project, write_server_info
create_project("test-proj", "Test", workbench_dir=tmp_workbench)
write_server_info("test-proj", pid=os.getpid(), port=9999, workbench_dir=tmp_workbench)
with patch.object(server, "_is_server_alive", return_value=True):
with patch.object(server, "_start_http_server", new_callable=AsyncMock) as mock_start:
result = await server.workbench_scaffold("test-proj", "Test")
mock_start.assert_not_called() # Should NOT start a new server
data = json.loads(result)
assert "9999" in data["url"]
@pytest.mark.asyncio
async def test_scaffold_replaces_dead_server(server, tmp_workbench):
"""If .server.json exists but server is dead, start a new one."""
from workbench.project import create_project, write_server_info
create_project("test-proj", "Test", workbench_dir=tmp_workbench)
write_server_info("test-proj", pid=99999, port=9999, workbench_dir=tmp_workbench)
with patch.object(server, "_is_server_alive", return_value=False):
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
result = await server.workbench_scaffold("test-proj", "Test")
data = json.loads(result)
assert "8070" in data["url"]
@pytest.mark.asyncio
async def test_list_empty(server):
result = await server.workbench_list()
data = json.loads(result)
assert data["projects"] == []
@pytest.mark.asyncio
async def test_log_writes_to_disk(server, tmp_workbench):
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
await server.workbench_scaffold("test-proj", "Test")
result = await server.workbench_log("test-proj", "R412 measured 1.05M")
data = json.loads(result)
assert data["ok"] is True
jsonl = (tmp_workbench / "test-proj" / "session.jsonl").read_text().strip()
assert "R412 measured 1.05M" in jsonl
@pytest.mark.asyncio
async def test_state_saves_to_disk(server, tmp_workbench):
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
await server.workbench_scaffold("test-proj", "Test")
state = json.dumps({"template": "<h1>Hello</h1>"})
result = await server.workbench_state("test-proj", state)
data = json.loads(result)
assert data["ok"] is True
saved = json.loads((tmp_workbench / "test-proj" / "state.json").read_text())
assert saved["template"] == "<h1>Hello</h1>"
@pytest.mark.asyncio
async def test_stop_cleans_up(server, tmp_workbench):
with patch.object(server, "_start_http_server", new_callable=AsyncMock, return_value=8070):
await server.workbench_scaffold("test-proj", "Test")
server._runners["test-proj"] = AsyncMock()
result = await server.workbench_stop("test-proj")
data = json.loads(result)
assert data["ok"] is True
assert not (tmp_workbench / "test-proj" / ".server.json").exists()
@pytest.mark.asyncio
async def test_reconnect_on_startup(tmp_workbench):
"""On init, server should find live servers from .server.json files."""
from workbench.project import create_project, write_server_info
create_project("live-proj", "Live", workbench_dir=tmp_workbench)
write_server_info("live-proj", pid=os.getpid(), port=8070, workbench_dir=tmp_workbench)
create_project("dead-proj", "Dead", workbench_dir=tmp_workbench)
write_server_info("dead-proj", pid=99999, port=8071, workbench_dir=tmp_workbench)
srv = WorkbenchServer(workbench_dir=tmp_workbench)
with patch.object(srv, "_is_server_alive", side_effect=lambda port: port == 8070):
await srv.reconnect_existing_servers()
assert "live-proj" in srv._active
assert srv._active["live-proj"]["port"] == 8070
assert "dead-proj" not in srv._active
# Dead server's .server.json should be cleaned up
assert not (tmp_workbench / "dead-proj" / ".server.json").exists()
```
- [ ] **Step 2: Run tests to verify they fail**
```bash
pytest tests/test_server.py -v
```
Expected: ImportError.
- [ ] **Step 3: Write server.py**
```python
"""MCP server — 6 workbench tools with HTTP/WS server management and persistence."""
from __future__ import annotations
import asyncio
import json
import os
import socket
from pathlib import Path
from typing import Optional
from aiohttp import web
from mcp.server.fastmcp import FastMCP
from workbench.project import (
create_project, project_exists, project_path,
append_log, read_log, list_projects, log_session_event,
write_server_info, read_server_info, clear_server_info,
WORKBENCH_DIR,
)
def _get_lan_ip() -> str:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("192.168.0.1", 80))
ip = s.getsockname()[0]
s.close()
return ip
except Exception:
return "127.0.0.1"
def _find_free_port(start: int = 8070) -> int:
for port in range(start, start + 100):
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind(("0.0.0.0", port))
s.close()
return port
except OSError:
continue
raise RuntimeError(f"No free port found in range {start}-{start + 99}")
class WorkbenchServer:
"""Core server logic — testable without MCP transport."""
def __init__(self, workbench_dir: Path = WORKBENCH_DIR):
self.workbench_dir = Path(workbench_dir)
self._active: dict[str, dict] = {} # name -> {"port": int, "runner": AppRunner, "ws_clients": set}
self._runners: dict[str, web.AppRunner] = {}
def _is_server_alive(self, port: int) -> bool:
"""Check if an HTTP server is responding on the given port."""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(1)
s.connect(("127.0.0.1", port))
s.close()
return True
except (OSError, ConnectionRefusedError):
return False
async def reconnect_existing_servers(self) -> None:
"""Scan for running servers from .server.json files. Called on MCP startup."""
if not self.workbench_dir.exists():
return
for d in self.workbench_dir.iterdir():
if not d.is_dir():
continue
name = d.name
info = read_server_info(name, workbench_dir=self.workbench_dir)
if info is None:
continue
port = info["port"]
if self._is_server_alive(port):
self._active[name] = {"port": port, "ws_clients": set()}
else:
clear_server_info(name, workbench_dir=self.workbench_dir)
async def _start_http_server(self, name: str) -> int:
"""Start an HTTP + WebSocket server for a project. Returns port."""
port = _find_free_port()
pdir = project_path(name, self.workbench_dir)
app = web.Application()
app["project_name"] = name
app["workbench_server"] = self
async def ws_handler(request):
ws = web.WebSocketResponse()
await ws.prepare(request)
proj = request.app["project_name"]
if proj in self._active:
self._active[proj]["ws_clients"].add(ws)
try:
async for msg in ws:
pass
finally:
if proj in self._active:
self._active[proj]["ws_clients"].discard(ws)
return ws
async def static_handler(request):
proj = request.app["project_name"]
path = request.match_info.get("path", "index.html") or "index.html"
file_path = project_path(proj, self.workbench_dir) / path
if not file_path.exists():
return web.Response(status=404, text="Not found")
return web.FileResponse(file_path)
app.router.add_get("/ws", ws_handler)
app.router.add_get("/{path:.*}", static_handler)
app.router.add_get("/", static_handler)
runner = web.AppRunner(app)
await runner.setup()
site = web.TCPSite(runner, "0.0.0.0", port)
await site.start()
self._runners[name] = runner
self._active[name] = {"port": port, "ws_clients": set()}
write_server_info(name, pid=os.getpid(), port=port, workbench_dir=self.workbench_dir)
return port
async def _broadcast_ws(self, name: str, message: dict) -> None:
if name not in self._active:
return
clients = self._active[name]["ws_clients"]
dead = set()
for ws in clients:
try:
await ws.send_json(message)
except Exception:
dead.add(ws)
clients -= dead
# --- MCP Tool implementations ---
async def workbench_scaffold(self, name: str, title: str, description: str = "") -> str:
pdir = project_path(name, self.workbench_dir)
create_project(name, title, description, workbench_dir=self.workbench_dir)
# Check for existing running server
info = read_server_info(name, workbench_dir=self.workbench_dir)
if info and self._is_server_alive(info["port"]):
port = info["port"]
if name not in self._active:
self._active[name] = {"port": port, "ws_clients": set()}
else:
if info:
clear_server_info(name, workbench_dir=self.workbench_dir)
port = await self._start_http_server(name)
ip = _get_lan_ip()
log_session_event(name, "session_start", workbench_dir=self.workbench_dir)
return json.dumps({"path": str(pdir), "url": f"http://{ip}:{port}"})
async def workbench_state(self, project: str, state: str) -> str:
if not project_exists(project, workbench_dir=self.workbench_dir):
return json.dumps({"error": f"Project '{project}' not found. Run workbench_scaffold first."})
state_obj = json.loads(state)
pdir = project_path(project, self.workbench_dir)
(pdir / "state.json").write_text(json.dumps(state_obj, indent=2))
await self._broadcast_ws(project, {"type": "state", "state": state_obj})
return json.dumps({"ok": True})
async def workbench_log(self, project: str, entry: str, data: str = "{}") -> str:
if not project_exists(project, workbench_dir=self.workbench_dir):
return json.dumps({"error": f"Project '{project}' not found."})
data_obj = json.loads(data) if data and data != "{}" else None
append_log(project, entry, data=data_obj, workbench_dir=self.workbench_dir)
await self._broadcast_ws(project, {"type": "log", "entry": entry})
return json.dumps({"ok": True})
async def workbench_read_log(self, project: str, tail: int = 20) -> str:
if not project_exists(project, workbench_dir=self.workbench_dir):
return json.dumps({"error": f"Project '{project}' not found."})
entries = read_log(project, tail=tail, workbench_dir=self.workbench_dir)
return json.dumps({"entries": entries})
async def workbench_list(self) -> str:
projects = list_projects(workbench_dir=self.workbench_dir)
for p in projects:
p["active"] = p["name"] in self._active
if p["active"]:
ip = _get_lan_ip()
port = self._active[p["name"]]["port"]
p["url"] = f"http://{ip}:{port}"
return json.dumps({"projects": projects})
async def workbench_stop(self, project: str) -> str:
if project not in self._active:
return json.dumps({"error": f"Project '{project}' is not running."})
# Log session end
if project_exists(project, workbench_dir=self.workbench_dir):
entries = read_log(project, tail=999999, workbench_dir=self.workbench_dir)
log_session_event(
project, "session_end",
workbench_dir=self.workbench_dir,
log_entries=len(entries),
)
# Shutdown HTTP server
if project in self._runners:
await self._runners[project].cleanup()
del self._runners[project]
clear_server_info(project, workbench_dir=self.workbench_dir)
del self._active[project]
return json.dumps({"ok": True})
def create_mcp_server(workbench_dir: Path = WORKBENCH_DIR) -> FastMCP:
"""Create the FastMCP instance with all workbench tools registered."""
srv = WorkbenchServer(workbench_dir=workbench_dir)
mcp = FastMCP(
"workbench",
instructions="Workbench — build interactive web pages served over LAN. Call workbench_scaffold first.",
)
@mcp.tool()
async def workbench_scaffold(name: str, title: str, description: str = "") -> str:
"""Create a workbench project and start the HTTP server. Returns the LAN URL to open in a browser. If the project already exists and its server is running, reattaches without starting a duplicate. Always safe to call — never creates duplicates."""
return await srv.workbench_scaffold(name, title, description)
@mcp.tool()
async def workbench_state(project: str, state: str) -> str:
"""Push a state update to the browser via WebSocket. The state is a JSON string — include 'template' (HTML string) to replace the page content, 'styles' (CSS string) to inject styles, and 'script' (JS string) to execute code. The AI has full control over the page."""
return await srv.workbench_state(project, state)
@mcp.tool()
async def workbench_log(project: str, entry: str, data: str = "{}") -> str:
"""Append a log entry to the session log. Shows in the browser log feed. entry: human-readable markdown string. data: optional JSON for the machine-readable log."""
return await srv.workbench_log(project, entry, data)
@mcp.tool()
async def workbench_read_log(project: str, tail: int = 20) -> str:
"""Read recent session log entries so the AI can resume a previous session."""
return await srv.workbench_read_log(project, tail)
@mcp.tool()
async def workbench_list() -> str:
"""List all workbench projects and whether their HTTP server is running."""
return await srv.workbench_list()
@mcp.tool()
async def workbench_stop(project: str) -> str:
"""Stop the HTTP server for a project and end the session."""
return await srv.workbench_stop(project)
return mcp
```
- [ ] **Step 4: Run tests**
```bash
pytest tests/test_server.py -v
```
Expected: All 9 tests pass.
- [ ] **Step 5: Run full suite**
```bash
pytest tests/ -v
```
Expected: All 18 tests pass (9 project + 9 server).
- [ ] **Step 6: Commit**
```bash
git add src/workbench/server.py tests/test_server.py
git commit -m "feat: MCP server with persistence — reattach to running servers on restart"
```
---
## Task 5: CLI Entry Point
**Files:**
- Create: `src/workbench/cli.py`
- [ ] **Step 1: Write cli.py**
```python
"""CLI entry point for workbench."""
from __future__ import annotations
import argparse
import asyncio
import sys
from workbench import __version__
def cmd_mcp(args):
"""Start the MCP server on stdio."""
from workbench.server import create_mcp_server
mcp = create_mcp_server()
mcp.run(transport="stdio")
def cmd_serve(args):
"""Serve a project without MCP."""
from workbench.server import WorkbenchServer
srv = WorkbenchServer()
async def run():
result = await srv.workbench_scaffold(args.name, args.name)
import json
data = json.loads(result)
print(f"Serving: {data['url']}")
try:
while True:
await asyncio.sleep(1)
except (KeyboardInterrupt, asyncio.CancelledError):
await srv.workbench_stop(args.name)
asyncio.run(run())
def cmd_list(args):
"""List all projects."""
from workbench.project import list_projects, read_server_info
projects = list_projects()
if not projects:
print("No projects found in ~/workbench/")
return
for p in projects:
info = read_server_info(p["name"])
status = f" (running on port {info['port']})" if info else ""
print(f" {p['name']}{status}")
def main():
parser = argparse.ArgumentParser(
prog="workbench",
description="MCP server for AI-driven interactive web pages",
)
parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
sub = parser.add_subparsers(dest="command")
sub.add_parser("mcp", help="Start MCP server (stdio transport)")
serve_parser = sub.add_parser("serve", help="Serve a project (standalone)")
serve_parser.add_argument("name", help="Project name")
sub.add_parser("list", help="List all projects")
sub.add_parser("help", help="Show help")
args = parser.parse_args()
commands = {
"mcp": cmd_mcp,
"serve": cmd_serve,
"list": cmd_list,
"help": lambda a: parser.print_help(),
}
if args.command is None:
parser.print_help()
sys.exit(0)
commands[args.command](args)
```
- [ ] **Step 2: Install and verify**
```bash
cd ~/bin/workbench-server
pip install --break-system-packages -e .
workbench --version
workbench help
workbench list
```
Expected: version "workbench 0.2.0", help shows subcommands, list says "No projects found".
- [ ] **Step 3: Commit**
```bash
git add src/workbench/cli.py
git commit -m "feat: CLI entry point — mcp, serve, list, help"
```
---
## Task 6: Clean Up Old Files
**Files:**
- Remove: `server.py` (root level — replaced by `src/workbench/server.py`)
- Remove: `scaffold.html` (root level — replaced by `src/workbench/scaffold.html`)
- [ ] **Step 1: Remove old files**
```bash
cd ~/bin/workbench-server
git rm server.py scaffold.html
```
- [ ] **Step 2: Run full test suite to confirm nothing breaks**
```bash
pytest tests/ -v
```
Expected: All 18 tests pass.
- [ ] **Step 3: Commit**
```bash
git commit -m "refactor: remove old root-level server.py and scaffold.html"
```
---
## Task 7: Docs — README, INSTALL.md, LICENSE
**Files:**
- Rewrite: `README.md`
- Create: `INSTALL.md`
- [ ] **Step 1: Rewrite README.md**
Public-facing. Desktop browser as display surface. Setup is "paste URL" or "read INSTALL.md". No mention of kitty or terminal splitting. Adapted from the kitty-workbench README pattern but for browser-based display.
Key sections: what it does (with ASCII art diagram), setup (clone + "tell AI to read INSTALL.md"), tools table, usage examples, project structure, requirements, FAQ.
- [ ] **Step 2: Write INSTALL.md**
AI-readable setup instructions. Simpler than kitty-workbench (no terminal detection needed):
0. Clone + pip install
1. Detect environment: Python version, platform, browser availability
2. Detect SSH usage: `$SSH_CONNECTION`, `~/.ssh/known_hosts`, ControlMaster config
3. Present options: SSH ControlMaster if applicable
4. Configure MCP for user's AI CLI (detect Claude Code, Gemini CLI, etc.)
5. Smoke test: `workbench list`
6. Write `~/workbench/START.md`
Include SSH ControlMaster setup (same as kitty-workbench INSTALL.md — `ControlMaster auto`, `ControlPersist yes`, sockets dir).
- [ ] **Step 3: Commit**
```bash
git add README.md INSTALL.md
git commit -m "docs: README and INSTALL.md — AI-guided setup"
```
---
## Task 8: Push to Gitea
- [ ] **Step 1: Run full test suite**
```bash
cd ~/bin/workbench-server
pytest tests/ -v
```
Expected: All 18 tests pass.
- [ ] **Step 2: Push**
```bash
cd ~/bin/workbench-server
gitea push
```
---
## Summary
| Task | What | Key files |
|------|------|-----------|
| 1 | Package scaffold + move files | pyproject.toml, __init__.py, __main__.py, LICENSE |
| 2 | Project management module | project.py, test_project.py |
| 3 | Scaffold HTML v2 | scaffold.html |
| 4 | MCP server with persistence | server.py, test_server.py |
| 5 | CLI entry point | cli.py |
| 6 | Clean up old files | remove server.py, scaffold.html from root |
| 7 | Docs | README.md, INSTALL.md |
| 8 | Push to Gitea | — |
Reference in New Issue View Git Blame Copy Permalink