ken/research-workbench
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c6d8d4dfe6ccde0e50048e6ddbcdc892df88bfd9
research-workbench/docs/plans/2026-05-26-research-workbench-phase1.md
T
Ken 4656af010e docs: implementation plans for research workbench (3 phases, 30 tasks) 
Phase 1: Infrastructure (8 tasks) - Dockerfile, amplifierd, FastAPI, auth, AG-UI adapter
Phase 2: Frontend (10 tasks) - React/TS/Vite, Tabler, CopilotKit headless, SSE streaming
Phase 3: Bundle + Browser + Artifacts (12 tasks) - Amplifier bundle, researcher agent,
  AI-generated visual artifacts (Claude builds visuals style), noVNC browser panel
2026-05-26 18:01:47 +00:00

1507 lines
46 KiB
Markdown
Raw Blame History

# Research Workbench — Phase 1: Infrastructure
> **Execution:** Use the subagent-driven-development workflow to implement this plan.
**Goal:** Build the Python backend (FastAPI + auth + AG-UI adapter + artifact API) and Docker container infrastructure for the research workbench.
**Architecture:** A single Docker container runs amplifierd (session daemon), FastAPI (serves frontend + proxies amplifierd via AG-UI protocol), playwright-cli (browser control), and Xvfb/noVNC (virtual display). FastAPI translates between amplifierd's SSE event format and the AG-UI protocol that CopilotKit expects. Tool results with `_meta.ui.resourceUri` (MCP Apps) are passed through so the frontend can render interactive HTML visualizations in sandboxed iframes. Auth is simple email + bcrypt password hash with session cookies.
**Tech Stack:** Python 3.13, FastAPI, uvicorn, httpx, bcrypt, ag-ui-protocol, Docker, Xvfb, x11vnc, websockify, noVNC
---
## Prerequisites
- The repo is at `/home/ken/workspace/research-workbench`
- It currently contains old car-help code (search.py, guide.py, sites.yaml, results/, static/, app.py) that will be replaced
- Keep: `DESIGN.md`, `README.md`, `.gitignore`, `.python-version`, `.git/`
---
### Task 1: Clean out old car-help files
**Files:**
- Delete: `search.py`, `guide.py`, `sites.yaml`, `app.py`
- Delete: `static/` directory, `results/` directory
- Delete: `Dockerfile`, `entrypoint.sh`, `pyproject.toml`, `uv.lock`
- Modify: `.gitignore`
**Step 1: Remove old files**
Run:
```bash
cd /home/ken/workspace/research-workbench
rm -f search.py guide.py sites.yaml app.py
rm -rf static/ results/
rm -f Dockerfile entrypoint.sh pyproject.toml uv.lock
```
**Step 2: Update .gitignore for new project structure**
Replace `.gitignore` with:
```gitignore
# Python
.venv/
__pycache__/
*.pyc
*.egg-info/
# Node / Frontend
frontend/node_modules/
frontend/dist/
# Runtime
artifacts/
# IDE
.idea/
.vscode/
*.swp
```
**Step 3: Create directory structure**
Run:
```bash
cd /home/ken/workspace/research-workbench
mkdir -p backend tests bundle/behaviors bundle/agents bundle/context frontend/src artifacts docs/plans
```
**Step 4: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add -A && git commit -m "chore: remove old car-help files, prepare new structure"
```
---
### Task 2: Create backend pyproject.toml
**Files:**
- Create: `backend/pyproject.toml`
**Step 1: Create the pyproject.toml**
Create `backend/pyproject.toml`:
```toml
[project]
name = "research-workbench"
version = "0.1.0"
description = "AI research workbench backend - FastAPI + AG-UI adapter for amplifierd"
requires-python = ">=3.13"
dependencies = [
"fastapi>=0.115.0",
"uvicorn[standard]>=0.34.0",
"httpx>=0.28.0",
"bcrypt>=4.2.0",
"python-multipart>=0.0.18",
"ag-ui-protocol>=0.1.18",
]
[dependency-groups]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.24",
"httpx>=0.28.0",
]
```
**Step 2: Initialize the venv and install deps**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv sync
```
Expected: Dependencies install successfully. A `uv.lock` file is created.
**Step 3: Verify ag-ui-protocol is importable**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -c "from ag_ui.core import RunAgentInput, EventType; print('ag-ui-protocol OK')"
```
Expected: `ag-ui-protocol OK`
**Step 4: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add backend/pyproject.toml backend/uv.lock && git commit -m "feat: add backend pyproject.toml with dependencies"
```
---
### Task 3: Auth module (email + bcrypt + session cookie)
**Files:**
- Create: `backend/auth.py`
- Create: `tests/test_auth.py`
**Step 1: Write the failing test**
Create `tests/test_auth.py`:
```python
"""Tests for auth module."""
import os
import sys
import pytest
from unittest.mock import patch
# Add backend to path so tests can import modules
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
from auth import verify_password, create_session_token, validate_session_token
class TestVerifyPassword:
"""Test bcrypt password verification."""
def test_correct_password(self):
import bcrypt
hashed = bcrypt.hashpw(b"testpass123", bcrypt.gensalt()).decode()
with patch.dict(os.environ, {"AUTH_PASS_HASH": hashed}):
assert verify_password("testpass123") is True
def test_wrong_password(self):
import bcrypt
hashed = bcrypt.hashpw(b"testpass123", bcrypt.gensalt()).decode()
with patch.dict(os.environ, {"AUTH_PASS_HASH": hashed}):
assert verify_password("wrongpassword") is False
def test_empty_password(self):
import bcrypt
hashed = bcrypt.hashpw(b"testpass123", bcrypt.gensalt()).decode()
with patch.dict(os.environ, {"AUTH_PASS_HASH": hashed}):
assert verify_password("") is False
class TestSessionToken:
"""Test session token creation and validation."""
def test_create_and_validate(self):
token = create_session_token("ken@example.com")
assert token is not None
assert len(token) > 20
def test_validate_valid_token(self):
token = create_session_token("ken@example.com")
email = validate_session_token(token)
assert email == "ken@example.com"
def test_validate_invalid_token(self):
email = validate_session_token("bogus-token-value")
assert email is None
def test_validate_empty_token(self):
email = validate_session_token("")
assert email is None
```
**Step 2: Run test to verify it fails**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_auth.py -v 2>&1 | head -20
```
Expected: FAIL with `ModuleNotFoundError: No module named 'auth'`
**Step 3: Write the implementation**
Create `backend/auth.py`:
```python
"""
Simple email + password authentication with session cookies.
Auth credentials come from environment variables:
AUTH_USER - the allowed email address
AUTH_PASS_HASH - bcrypt hash of the password
Session tokens are stored in-memory (single container, single user).
Token format: random hex string mapped to email in a dict.
"""
import os
import secrets
import bcrypt
AUTH_USER: str = os.environ.get("AUTH_USER", "admin@localhost")
AUTH_PASS_HASH: str = os.environ.get("AUTH_PASS_HASH", "")
# In-memory session store: token -> email
_sessions: dict[str, str] = {}
def verify_password(password: str) -> bool:
"""Check a plaintext password against the AUTH_PASS_HASH env var."""
stored_hash = os.environ.get("AUTH_PASS_HASH", AUTH_PASS_HASH)
if not stored_hash or not password:
return False
try:
return bcrypt.checkpw(password.encode("utf-8"), stored_hash.encode("utf-8"))
except (ValueError, TypeError):
return False
def create_session_token(email: str) -> str:
"""Create a new session token for the given email."""
token = secrets.token_hex(32)
_sessions[token] = email
return token
def validate_session_token(token: str) -> str | None:
"""Return the email for a valid token, or None."""
if not token:
return None
return _sessions.get(token)
def invalidate_session_token(token: str) -> None:
"""Remove a session token."""
_sessions.pop(token, None)
```
**Step 4: Run tests to verify they pass**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_auth.py -v
```
Expected: All 6 tests PASS.
**Step 5: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add backend/auth.py tests/test_auth.py && git commit -m "feat: auth module with bcrypt password verification and session tokens"
```
---
### Task 4: Artifacts module (per-session CRUD)
**Files:**
- Create: `backend/artifacts.py`
- Create: `tests/test_artifacts.py`
**Step 1: Write the failing test**
Create `tests/test_artifacts.py`:
```python
"""Tests for artifacts module."""
import os
import sys
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
from artifacts import list_artifacts, get_artifact, save_artifact
@pytest.fixture(autouse=True)
def temp_artifacts(tmp_path, monkeypatch):
"""Use a temp directory for artifacts during tests."""
monkeypatch.setattr("artifacts.ARTIFACTS_DIR", tmp_path)
return tmp_path
class TestArtifacts:
def test_list_empty_session(self):
result = list_artifacts("session-1")
assert result == []
def test_save_and_list(self, temp_artifacts):
save_artifact("session-1", "guide.md", "# My Guide\n\nContent here.")
result = list_artifacts("session-1")
assert len(result) == 1
assert result[0]["name"] == "guide.md"
assert result[0]["session_id"] == "session-1"
def test_save_and_get(self, temp_artifacts):
save_artifact("session-1", "report.md", "# Report\n\nFindings.")
content = get_artifact("session-1", "report.md")
assert content == "# Report\n\nFindings."
def test_get_nonexistent(self):
content = get_artifact("no-session", "no-file.md")
assert content is None
def test_multiple_artifacts(self, temp_artifacts):
save_artifact("s1", "a.md", "A")
save_artifact("s1", "b.md", "B")
save_artifact("s2", "c.md", "C")
assert len(list_artifacts("s1")) == 2
assert len(list_artifacts("s2")) == 1
def test_overwrite_artifact(self, temp_artifacts):
save_artifact("s1", "doc.md", "v1")
save_artifact("s1", "doc.md", "v2")
assert get_artifact("s1", "doc.md") == "v2"
assert len(list_artifacts("s1")) == 1
```
**Step 2: Run test to verify it fails**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_artifacts.py -v 2>&1 | head -10
```
Expected: FAIL with `ModuleNotFoundError: No module named 'artifacts'`
**Step 3: Write the implementation**
Create `backend/artifacts.py`:
```python
"""
Per-session artifact storage.
Artifacts are markdown files stored at artifacts/<session_id>/<name>.md
Written by the researcher agent via tool-filesystem, served by FastAPI.
"""
from pathlib import Path
ARTIFACTS_DIR = Path("/app/artifacts")
def list_artifacts(session_id: str) -> list[dict]:
"""List all artifacts for a session."""
session_dir = ARTIFACTS_DIR / session_id
if not session_dir.is_dir():
return []
artifacts = []
for f in sorted(session_dir.iterdir()):
if f.is_file():
artifacts.append({
"name": f.name,
"session_id": session_id,
"size": f.stat().st_size,
})
return artifacts
def get_artifact(session_id: str, name: str) -> str | None:
"""Get the content of a specific artifact."""
path = ARTIFACTS_DIR / session_id / name
if not path.is_file():
return None
return path.read_text(encoding="utf-8")
def save_artifact(session_id: str, name: str, content: str) -> Path:
"""Save (or overwrite) an artifact."""
session_dir = ARTIFACTS_DIR / session_id
session_dir.mkdir(parents=True, exist_ok=True)
path = session_dir / name
path.write_text(content, encoding="utf-8")
return path
```
**Step 4: Run tests to verify they pass**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_artifacts.py -v
```
Expected: All 6 tests PASS.
**Step 5: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add backend/artifacts.py tests/test_artifacts.py && git commit -m "feat: artifact storage module with per-session CRUD"
```
---
### Task 5: AG-UI adapter (amplifierd SSE -> AG-UI events, MCP App metadata passthrough)
This is the critical integration piece. It translates amplifierd's SSE event format into AG-UI protocol events that CopilotKit understands. It also passes through `_meta.ui` and `structuredContent` from tool results so the frontend can render MCP App iframes for interactive visualizations.
**amplifierd SSE events** (from `GET /events?session=X`):
- `content_block:start` — `{block_type: "text"|"tool_use", block_index, tool_id?, tool_name?}`
- `content_block:delta` — `{token?, block_type?, content?, block_index}`
- `content_block:end` — `{block_index}`
- `tool:result` — `{tool_call_id, tool_name, content}`
- `orchestrator:complete` — `{}`
**AG-UI events** (what CopilotKit frontend expects):
- `RUN_STARTED`, `TEXT_MESSAGE_START/CONTENT/END`, `TOOL_CALL_START/ARGS/END/RESULT`, `RUN_FINISHED`
**Files:**
- Create: `backend/agui_adapter.py`
- Create: `tests/test_agui_adapter.py`
**Step 1: Write the failing test**
Create `tests/test_agui_adapter.py`:
```python
"""Tests for the AG-UI adapter that translates amplifierd SSE to AG-UI events."""
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
from agui_adapter import AmplifierdEventTranslator
class TestTextBlocks:
"""Test text content block translation."""
def test_text_block_start(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
event = {
"event_type": "content_block:start",
"data": {"block_type": "text", "block_index": 0},
}
ag_events = list(translator.translate(event))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TEXT_MESSAGE_START"
assert ag_events[0]["role"] == "assistant"
assert "messageId" in ag_events[0]
def test_text_delta(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
# Start text block first
translator.translate({
"event_type": "content_block:start",
"data": {"block_type": "text", "block_index": 0},
})
ag_events = list(translator.translate({
"event_type": "content_block:delta",
"data": {"token": "Hello ", "block_index": 0},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TEXT_MESSAGE_CONTENT"
assert ag_events[0]["delta"] == "Hello "
def test_text_block_end(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
translator.translate({
"event_type": "content_block:start",
"data": {"block_type": "text", "block_index": 0},
})
ag_events = list(translator.translate({
"event_type": "content_block:end",
"data": {"block_index": 0},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TEXT_MESSAGE_END"
class TestToolBlocks:
"""Test tool call translation."""
def test_tool_use_start(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
ag_events = list(translator.translate({
"event_type": "content_block:start",
"data": {
"block_type": "tool_use",
"block_index": 1,
"tool_id": "tool_123",
"tool_name": "bash",
},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TOOL_CALL_START"
assert ag_events[0]["toolCallName"] == "bash"
assert ag_events[0]["toolCallId"] == "tool_123"
def test_tool_use_delta(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
translator.translate({
"event_type": "content_block:start",
"data": {
"block_type": "tool_use",
"block_index": 1,
"tool_id": "tool_123",
"tool_name": "bash",
},
})
ag_events = list(translator.translate({
"event_type": "content_block:delta",
"data": {
"block_type": "tool_use",
"content": '{"command": "ls"}',
"block_index": 1,
},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TOOL_CALL_ARGS"
assert ag_events[0]["delta"] == '{"command": "ls"}'
def test_tool_use_end(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
translator.translate({
"event_type": "content_block:start",
"data": {
"block_type": "tool_use",
"block_index": 1,
"tool_id": "tool_123",
"tool_name": "bash",
},
})
ag_events = list(translator.translate({
"event_type": "content_block:end",
"data": {"block_index": 1},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TOOL_CALL_END"
def test_tool_result(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
ag_events = list(translator.translate({
"event_type": "tool:result",
"data": {
"tool_call_id": "tool_123",
"tool_name": "bash",
"content": "file1.txt\nfile2.txt",
},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TOOL_CALL_RESULT"
assert ag_events[0]["toolCallId"] == "tool_123"
def test_tool_result_with_mcp_app_metadata(self):
"""Tool results with _meta.ui.resourceUri pass through UI metadata for MCP Apps."""
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
ag_events = list(translator.translate({
"event_type": "tool:result",
"data": {
"tool_call_id": "tool_456",
"tool_name": "show_chart",
"content": "Chart rendered with 5 data points",
"_meta": {
"ui": {
"resourceUri": "app://chart",
"csp": "default-src 'self'; script-src 'unsafe-inline' cdn.jsdelivr.net",
}
},
"structuredContent": {
"chartType": "bar",
"labels": ["Q1", "Q2", "Q3"],
"datasets": [{"label": "Sales", "data": [10, 20, 30]}],
},
},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "TOOL_CALL_RESULT"
assert ag_events[0]["toolCallId"] == "tool_456"
# _meta.ui must be passed through for the frontend MCPAppRenderer
assert ag_events[0]["_meta"]["ui"]["resourceUri"] == "app://chart"
# structuredContent must be passed through for the iframe data
assert ag_events[0]["structuredContent"]["chartType"] == "bar"
def test_tool_result_without_mcp_app_metadata(self):
"""Tool results without _meta.ui should NOT have _meta or structuredContent keys."""
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
ag_events = list(translator.translate({
"event_type": "tool:result",
"data": {
"tool_call_id": "tool_789",
"tool_name": "bash",
"content": "hello world",
},
}))
assert len(ag_events) == 1
assert "_meta" not in ag_events[0]
assert "structuredContent" not in ag_events[0]
class TestLifecycle:
"""Test lifecycle event translation."""
def test_orchestrator_complete(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
ag_events = list(translator.translate({
"event_type": "orchestrator:complete",
"data": {},
}))
assert len(ag_events) == 1
assert ag_events[0]["type"] == "RUN_FINISHED"
assert ag_events[0]["threadId"] == "thread-1"
assert ag_events[0]["runId"] == "run-1"
def test_unknown_event_ignored(self):
translator = AmplifierdEventTranslator(run_id="run-1", thread_id="thread-1")
ag_events = list(translator.translate({
"event_type": "some:unknown:event",
"data": {"foo": "bar"},
}))
assert ag_events == []
```
**Step 2: Run test to verify it fails**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_agui_adapter.py -v 2>&1 | head -10
```
Expected: FAIL with `ModuleNotFoundError: No module named 'agui_adapter'`
**Step 3: Write the implementation**
Create `backend/agui_adapter.py`:
```python
"""
AG-UI protocol adapter for amplifierd.
Translates amplifierd's SSE events into AG-UI protocol events that CopilotKit
can consume. This is the critical bridge between the two systems.
amplifierd events (from GET /events?session=X):
content_block:start {block_type: "text"|"tool_use", block_index, tool_id?, tool_name?}
content_block:delta {token?, block_type?, content?, block_index}
content_block:end {block_index}
tool:result {tool_call_id, tool_name, content}
orchestrator:complete {}
AG-UI events (sent to CopilotKit frontend):
RUN_STARTED {threadId, runId}
TEXT_MESSAGE_START {messageId, role}
TEXT_MESSAGE_CONTENT {messageId, delta}
TEXT_MESSAGE_END {messageId}
TOOL_CALL_START {toolCallId, toolCallName, parentMessageId?}
TOOL_CALL_ARGS {toolCallId, delta}
TOOL_CALL_END {toolCallId}
TOOL_CALL_RESULT {toolCallId, content, role, _meta?, structuredContent?}
RUN_FINISHED {threadId, runId}
MCP App passthrough:
When a tool:result includes _meta.ui.resourceUri, the TOOL_CALL_RESULT event
preserves _meta and structuredContent so the frontend can render an MCP App
iframe instead of a plain ToolCallCard.
"""
import uuid
class AmplifierdEventTranslator:
"""Stateful translator from amplifierd SSE events to AG-UI events.
Tracks active content blocks to correctly pair start/end events and
assign consistent message IDs across the stream.
"""
def __init__(self, run_id: str, thread_id: str):
self.run_id = run_id
self.thread_id = thread_id
# Track active blocks: block_index -> {type, message_id, tool_id}
self._blocks: dict[int, dict] = {}
# Current assistant message ID (for text blocks)
self._current_message_id: str | None = None
def translate(self, event: dict) -> list[dict]:
"""Translate a single amplifierd event into zero or more AG-UI events."""
event_type = event.get("event_type", "")
data = event.get("data", {})
if event_type == "content_block:start":
return self._handle_block_start(data)
elif event_type == "content_block:delta":
return self._handle_block_delta(data)
elif event_type == "content_block:end":
return self._handle_block_end(data)
elif event_type == "tool:result":
return self._handle_tool_result(data)
elif event_type == "orchestrator:complete":
return self._handle_complete()
else:
return []
def _handle_block_start(self, data: dict) -> list[dict]:
block_type = data.get("block_type", "text")
block_index = data.get("block_index", 0)
if block_type == "tool_use":
tool_id = data.get("tool_id", str(uuid.uuid4()))
tool_name = data.get("tool_name", "unknown")
self._blocks[block_index] = {
"type": "tool_use",
"tool_id": tool_id,
"tool_name": tool_name,
}
return [{
"type": "TOOL_CALL_START",
"toolCallId": tool_id,
"toolCallName": tool_name,
"parentMessageId": self._current_message_id,
}]
else:
message_id = str(uuid.uuid4())
self._current_message_id = message_id
self._blocks[block_index] = {
"type": "text",
"message_id": message_id,
}
return [{
"type": "TEXT_MESSAGE_START",
"messageId": message_id,
"role": "assistant",
}]
def _handle_block_delta(self, data: dict) -> list[dict]:
block_index = data.get("block_index", 0)
block_info = self._blocks.get(block_index)
if block_info is None:
# Delta without a matching start - handle gracefully
if data.get("block_type") == "tool_use":
tool_id = data.get("tool_id", "unknown")
return [{
"type": "TOOL_CALL_ARGS",
"toolCallId": tool_id,
"delta": data.get("content", ""),
}]
return []
if block_info["type"] == "tool_use":
return [{
"type": "TOOL_CALL_ARGS",
"toolCallId": block_info["tool_id"],
"delta": data.get("content", ""),
}]
else:
token = data.get("token", "")
if not token:
return []
return [{
"type": "TEXT_MESSAGE_CONTENT",
"messageId": block_info["message_id"],
"delta": token,
}]
def _handle_block_end(self, data: dict) -> list[dict]:
block_index = data.get("block_index", 0)
block_info = self._blocks.pop(block_index, None)
if block_info is None:
return []
if block_info["type"] == "tool_use":
return [{
"type": "TOOL_CALL_END",
"toolCallId": block_info["tool_id"],
}]
else:
return [{
"type": "TEXT_MESSAGE_END",
"messageId": block_info["message_id"],
}]
def _handle_tool_result(self, data: dict) -> list[dict]:
event: dict = {
"type": "TOOL_CALL_RESULT",
"toolCallId": data.get("tool_call_id", "unknown"),
"content": data.get("content", ""),
"role": "tool",
}
# Pass through MCP App metadata if present.
# When a tool result includes _meta.ui.resourceUri, the frontend
# renders an MCPAppRenderer (sandboxed iframe) instead of a ToolCallCard.
if "_meta" in data:
event["_meta"] = data["_meta"]
if "structuredContent" in data:
event["structuredContent"] = data["structuredContent"]
return [event]
def _handle_complete(self) -> list[dict]:
return [{
"type": "RUN_FINISHED",
"threadId": self.thread_id,
"runId": self.run_id,
}]
```
**Step 4: Run tests to verify they pass**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_agui_adapter.py -v
```
Expected: All 10 tests PASS.
**Step 5: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add backend/agui_adapter.py tests/test_agui_adapter.py && git commit -m "feat: AG-UI adapter translating amplifierd SSE events to AG-UI protocol (with MCP App metadata passthrough)"
```
---
### Task 6: FastAPI application (app.py)
The main FastAPI app: auth endpoints, AG-UI/CopilotKit SSE endpoint, artifact API, MCP App template serving, session proxy to amplifierd, and static file serving.
**Files:**
- Create: `backend/app.py`
- Create: `tests/test_app.py`
**Step 1: Write the failing test**
Create `tests/test_app.py`:
```python
"""Integration tests for the FastAPI app."""
import os
import sys
import bcrypt
import pytest
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
# Set auth env vars before importing app
TEST_HASH = bcrypt.hashpw(b"testpass", bcrypt.gensalt()).decode()
os.environ["AUTH_USER"] = "test@example.com"
os.environ["AUTH_PASS_HASH"] = TEST_HASH
from fastapi.testclient import TestClient
from app import app
client = TestClient(app)
class TestHealthEndpoint:
def test_health(self):
resp = client.get("/api/health")
assert resp.status_code == 200
assert resp.json()["status"] == "ok"
class TestAuthEndpoints:
def test_login_success(self):
resp = client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "testpass",
})
assert resp.status_code == 200
data = resp.json()
assert "token" in data
assert "session" in resp.cookies
def test_login_wrong_password(self):
resp = client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "wrongpass",
})
assert resp.status_code == 401
def test_login_wrong_email(self):
resp = client.post("/api/auth/login", json={
"email": "wrong@example.com",
"password": "testpass",
})
assert resp.status_code == 401
def test_me_unauthenticated(self):
resp = client.get("/api/auth/me")
assert resp.status_code == 401
def test_me_authenticated(self):
login_resp = client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "testpass",
})
token = login_resp.json()["token"]
resp = client.get("/api/auth/me", cookies={"session": token})
assert resp.status_code == 200
assert resp.json()["email"] == "test@example.com"
def test_logout(self):
login_resp = client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "testpass",
})
token = login_resp.json()["token"]
resp = client.post("/api/auth/logout", cookies={"session": token})
assert resp.status_code == 200
# Token should now be invalid
resp2 = client.get("/api/auth/me", cookies={"session": token})
assert resp2.status_code == 401
class TestArtifactEndpoints:
def _auth_cookies(self) -> dict:
resp = client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "testpass",
})
return {"session": resp.json()["token"]}
def test_list_artifacts_empty(self, tmp_path, monkeypatch):
monkeypatch.setattr("artifacts.ARTIFACTS_DIR", tmp_path)
cookies = self._auth_cookies()
resp = client.get("/api/artifacts/test-session", cookies=cookies)
assert resp.status_code == 200
assert resp.json() == []
def test_list_artifacts_unauthenticated(self):
resp = client.get("/api/artifacts/test-session")
assert resp.status_code == 401
class TestMCPAppsEndpoint:
"""Test serving MCP App HTML templates."""
def _auth_cookies(self) -> dict:
resp = client.post("/api/auth/login", json={
"email": "test@example.com",
"password": "testpass",
})
return {"session": resp.json()["token"]}
def test_get_app_not_found(self):
cookies = self._auth_cookies()
resp = client.get("/api/apps/nonexistent", cookies=cookies)
assert resp.status_code == 404
def test_get_app_unauthenticated(self):
resp = client.get("/api/apps/chart")
assert resp.status_code == 401
```
**Step 2: Run test to verify it fails**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_app.py -v 2>&1 | head -10
```
Expected: FAIL with `ModuleNotFoundError: No module named 'app'`
**Step 3: Write the implementation**
Create `backend/app.py` with the full FastAPI application. This is the longest file. Key sections:
- Auth endpoints (`/api/auth/login`, `/api/auth/me`, `/api/auth/logout`)
- Session proxy to amplifierd (`/api/sessions/*`)
- AG-UI endpoint for CopilotKit (`/api/copilotkit`) - receives RunAgentInput, streams AG-UI events
- Artifact API (`/api/artifacts/*`)
- Static file serving for built frontend
```python
"""
Research Workbench - FastAPI application.
Serves:
/api/health - health check
/api/auth/* - login, logout, me
/api/sessions/* - proxy to amplifierd session API
/api/copilotkit - AG-UI protocol endpoint for CopilotKit
/api/artifacts/* - per-session artifact CRUD
/api/apps/{name} - MCP App HTML templates (for sandboxed iframes)
/ - built frontend static files
"""
import json
import os
import uuid
from contextlib import asynccontextmanager
from pathlib import Path
import httpx
from fastapi import Cookie, FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from agui_adapter import AmplifierdEventTranslator
from artifacts import get_artifact, list_artifacts
from auth import (
create_session_token,
invalidate_session_token,
validate_session_token,
verify_password,
)
AMPLIFIERD_URL = os.environ.get("AMPLIFIERD_URL", "http://localhost:8410")
AUTH_USER = os.environ.get("AUTH_USER", "admin@localhost")
BUNDLE_NAME = os.environ.get("BUNDLE_NAME", "research-workbench")
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.http_client = httpx.AsyncClient(base_url=AMPLIFIERD_URL, timeout=30.0)
yield
await app.state.http_client.aclose()
app = FastAPI(title="research-workbench", version="0.1.0", lifespan=lifespan)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
)
# --- Auth helpers ---
def _require_auth(session: str | None) -> str:
email = validate_session_token(session) if session else None
if email is None:
raise HTTPException(status_code=401, detail="Not authenticated")
return email
# --- Health ---
@app.get("/api/health")
async def health():
return {"status": "ok"}
# --- Auth endpoints ---
@app.post("/api/auth/login")
async def login(request: Request):
body = await request.json()
email = body.get("email", "")
password = body.get("password", "")
if email != AUTH_USER or not verify_password(password):
return JSONResponse({"error": "Invalid credentials"}, status_code=401)
token = create_session_token(email)
response = JSONResponse({"token": token, "email": email})
response.set_cookie(key="session", value=token, httponly=True,
samesite="lax", max_age=86400 * 7)
return response
@app.get("/api/auth/me")
async def me(session: str | None = Cookie(default=None)):
email = validate_session_token(session) if session else None
if email is None:
return JSONResponse({"error": "Not authenticated"}, status_code=401)
return {"email": email}
@app.post("/api/auth/logout")
async def logout(session: str | None = Cookie(default=None)):
if session:
invalidate_session_token(session)
response = JSONResponse({"status": "ok"})
response.delete_cookie("session")
return response
# --- Session proxy (to amplifierd) ---
@app.get("/api/sessions")
async def list_sessions(session: str | None = Cookie(default=None)):
_require_auth(session)
resp = await app.state.http_client.get("/sessions")
return JSONResponse(resp.json(), status_code=resp.status_code)
@app.post("/api/sessions")
async def create_session(request: Request, session: str | None = Cookie(default=None)):
_require_auth(session)
body = await request.json()
if "bundle_name" not in body:
body["bundle_name"] = BUNDLE_NAME
resp = await app.state.http_client.post("/sessions", json=body)
return JSONResponse(resp.json(), status_code=resp.status_code)
@app.get("/api/sessions/{session_id}")
async def get_session(session_id: str, session: str | None = Cookie(default=None)):
_require_auth(session)
resp = await app.state.http_client.get(f"/sessions/{session_id}")
return JSONResponse(resp.json(), status_code=resp.status_code)
@app.delete("/api/sessions/{session_id}")
async def delete_session(session_id: str, session: str | None = Cookie(default=None)):
_require_auth(session)
resp = await app.state.http_client.delete(f"/sessions/{session_id}")
return JSONResponse({"status": "deleted"}, status_code=resp.status_code)
@app.get("/api/sessions/{session_id}/transcript")
async def get_transcript(session_id: str, session: str | None = Cookie(default=None)):
_require_auth(session)
resp = await app.state.http_client.get(f"/sessions/{session_id}/transcript")
return JSONResponse(resp.json(), status_code=resp.status_code)
# --- AG-UI endpoint (CopilotKit connects here) ---
@app.post("/api/copilotkit")
async def copilotkit_endpoint(request: Request, session: str | None = Cookie(default=None)):
"""AG-UI protocol endpoint.
CopilotKit sends RunAgentInput (messages, threadId, runId).
We map threadId to an amplifierd session, execute the prompt,
and stream back AG-UI events translated from amplifierd's SSE.
"""
_require_auth(session)
body = await request.json()
thread_id = body.get("threadId", str(uuid.uuid4()))
run_id = body.get("runId", str(uuid.uuid4()))
messages = body.get("messages", [])
# Extract latest user message as prompt
prompt = ""
for msg in reversed(messages):
if msg.get("role") == "user" and msg.get("content"):
prompt = msg["content"]
break
if not prompt:
return JSONResponse({"error": "No user message found"}, status_code=400)
# Use threadId as amplifierd session_id
amp_session_id = thread_id
client = app.state.http_client
# Create session if it doesn't exist
check = await client.get(f"/sessions/{amp_session_id}")
if check.status_code == 404:
await client.post("/sessions", json={
"session_id": amp_session_id,
"bundle_name": BUNDLE_NAME,
})
# Fire execute/stream
exec_resp = await client.post(
f"/sessions/{amp_session_id}/execute/stream",
json={"prompt": prompt},
)
if exec_resp.status_code != 202:
return JSONResponse(
{"error": "Failed to start execution", "detail": exec_resp.text},
status_code=502,
)
async def event_generator():
translator = AmplifierdEventTranslator(run_id=run_id, thread_id=thread_id)
yield _sse({"type": "RUN_STARTED", "threadId": thread_id, "runId": run_id})
async with httpx.AsyncClient(base_url=AMPLIFIERD_URL, timeout=None) as sse_client:
async with sse_client.stream(
"GET", "/events", params={"session": amp_session_id},
) as response:
buffer = ""
async for chunk in response.aiter_text():
buffer += chunk
while "\n\n" in buffer:
raw_event, buffer = buffer.split("\n\n", 1)
parsed = _parse_sse(raw_event)
if parsed is None:
continue
for ag_event in translator.translate(parsed):
yield _sse(ag_event)
if parsed.get("event_type") == "orchestrator:complete":
return
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
)
def _sse(event: dict) -> str:
return f"data: {json.dumps(event)}\n\n"
def _parse_sse(raw: str) -> dict | None:
event_type = None
data_lines = []
for line in raw.strip().split("\n"):
if line.startswith("event:"):
event_type = line[6:].strip()
elif line.startswith("data:"):
data_lines.append(line[5:].strip())
if not data_lines:
return None
try:
data = json.loads("\n".join(data_lines))
except json.JSONDecodeError:
return None
return {"event_type": event_type or data.get("event_type", ""), "data": data}
# --- Artifacts ---
@app.get("/api/artifacts/{session_id}")
async def api_list_artifacts(session_id: str, session: str | None = Cookie(default=None)):
_require_auth(session)
return JSONResponse(list_artifacts(session_id))
@app.get("/api/artifacts/{session_id}/{name}")
async def api_get_artifact(session_id: str, name: str,
session: str | None = Cookie(default=None)):
_require_auth(session)
content = get_artifact(session_id, name)
if content is None:
return JSONResponse({"error": "Not found"}, status_code=404)
return JSONResponse({"name": name, "session_id": session_id, "content": content})
# --- MCP App templates (served to sandboxed iframes) ---
APPS_DIR = Path(__file__).parent.parent / "bundle" / "apps"
@app.get("/api/apps/{app_name}")
async def get_mcp_app(app_name: str, session: str | None = Cookie(default=None)):
"""Serve MCP App HTML templates for inline interactive visualizations.
MCP Apps are self-contained HTML files rendered in sandboxed iframes.
Tool results with _meta.ui.resourceUri reference these apps.
"""
_require_auth(session)
# Normalize: allow "chart" or "chart.html"
if not app_name.endswith(".html"):
app_name += ".html"
app_path = APPS_DIR / app_name
if not app_path.is_file():
return JSONResponse({"error": f"App '{app_name}' not found"}, status_code=404)
html = app_path.read_text(encoding="utf-8")
return HTMLResponse(content=html)
# --- Static files (built frontend) - MUST be last ---
_static_dir = Path(__file__).parent.parent / "frontend" / "dist"
if _static_dir.is_dir():
app.mount("/", StaticFiles(directory=str(_static_dir), html=True), name="frontend")
```
**Step 4: Run tests to verify they pass**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/test_app.py -v
```
Expected: All 11 tests PASS. (Session proxy tests that call amplifierd will fail without amplifierd running -- that's expected. The tests above only cover auth, health, and artifact endpoints.)
**Step 5: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add backend/app.py tests/test_app.py && git commit -m "feat: FastAPI app with auth, AG-UI endpoint, session proxy, artifact API"
```
---
### Task 7: Dockerfile
**Files:**
- Create: `Dockerfile`
**Step 1: Create the Dockerfile**
Create `Dockerfile`:
```dockerfile
FROM python:3.13-slim
# System deps: display, VNC, browser libs, Node.js for playwright-cli
RUN apt-get update -qq && apt-get install -y -qq --no-install-recommends \
xvfb x11vnc websockify novnc curl \
libnss3 libatk1.0-0 libatk-bridge2.0-0 libcups2 libdrm2 \
libxkbcommon0 libxcomposite1 libxdamage1 libxrandr2 libgbm1 \
libpango-1.0-0 libcairo2 libasound2 libxshmfence1 libgtk-3-0 \
fonts-liberation && \
rm -rf /var/lib/apt/lists/*
# Install Node.js (for playwright-cli)
RUN curl -fsSL https://deb.nodesource.com/setup_22.x | bash - && \
apt-get install -y nodejs && \
rm -rf /var/lib/apt/lists/*
# Install playwright-cli globally
RUN npm install -g @anthropic-ai/playwright-cli
# Install uv
RUN pip install uv
# Backend deps
COPY backend/pyproject.toml backend/uv.lock /app/backend/
WORKDIR /app/backend
RUN uv sync --frozen
# Install Playwright browser
RUN uv run playwright install chromium
# Install amplifierd
RUN uv pip install amplifierd
# Copy backend source
COPY backend/*.py /app/backend/
# Copy bundle
COPY bundle/ /app/bundle/
# Copy built frontend (build before docker build)
COPY frontend/dist/ /app/frontend/dist/
# Copy entrypoint
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
# Create artifacts directory
RUN mkdir -p /app/artifacts
WORKDIR /app
EXPOSE 8080 6080
ENV DISPLAY=:99
ENV AMPLIFIERD_URL=http://localhost:8410
ENV BUNDLE_NAME=research-workbench
ENTRYPOINT ["/entrypoint.sh"]
```
**Step 2: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add Dockerfile && git commit -m "feat: Dockerfile with all-in-one container"
```
---
### Task 8: Entrypoint script
**Files:**
- Create: `entrypoint.sh`
**Step 1: Create the entrypoint**
Create `entrypoint.sh`:
```bash
#!/bin/bash
set -e
export DISPLAY=:99
echo "=== Starting Research Workbench ==="
# 1. Virtual display
Xvfb :99 -screen 0 1280x720x24 -ac &
sleep 1
echo "[OK] Xvfb display :99"
# 2. VNC server
x11vnc -display :99 -forever -nopw -listen 0.0.0.0 -rfbport 5900 -shared 2>/dev/null &
sleep 1
echo "[OK] x11vnc on :5900"
# 3. noVNC (web VNC client)
websockify --web=/usr/share/novnc/ 6080 localhost:5900 &
sleep 1
echo "[OK] noVNC on :6080"
# 4. amplifierd (session daemon)
cd /app
amplifierd --port 8410 --bundle /app/bundle/bundle.md &
AMPLIFIERD_PID=$!
# Wait for amplifierd to be ready
echo "Waiting for amplifierd..."
for i in $(seq 1 30); do
if curl -s http://localhost:8410/health > /dev/null 2>&1; then
echo "[OK] amplifierd on :8410"
break
fi
sleep 1
done
# 5. FastAPI (web frontend + API)
cd /app/backend
uv run python -m uvicorn app:app --host 0.0.0.0 --port 8080 --log-level info &
sleep 2
echo "[OK] FastAPI on :8080"
echo "============================================="
echo " Research Workbench: http://localhost:8080"
echo " Browser VNC: http://localhost:6080"
echo "============================================="
# Keep container running
if [ $# -eq 0 ]; then
wait $AMPLIFIERD_PID
else
exec "$@"
fi
```
**Step 2: Commit**
Run:
```bash
cd /home/ken/workspace/research-workbench
git add entrypoint.sh && git commit -m "feat: entrypoint.sh starting all services"
```
---
## Phase 1 Complete
At this point you have:
- Clean repo with new directory structure
- Backend with auth, artifacts, AG-UI adapter (with MCP App metadata passthrough), and FastAPI app
- MCP App template serving endpoint (`/api/apps/{name}`)
- Tests for auth, artifacts, AG-UI adapter (including MCP App passthrough), and app integration
- Dockerfile and entrypoint for the all-in-one container
- No frontend yet (that's Phase 2)
**Verify everything passes:**
Run:
```bash
cd /home/ken/workspace/research-workbench/backend
uv run python -m pytest ../tests/ -v
```
Expected: All tests pass.
Reference in New Issue View Git Blame Copy Permalink