12 KiB
src/api_hooks.py & src/api_hook_client.py — Hook API
Top | Architecture | Testing
Overview
The Hook API is the bridge between external automation and the running app. It exposes Manual Slop's internal state over HTTP on 127.0.0.1:8999, allowing tests, scripts, and external agents to drive the GUI without direct Python access.
Two files:
src/api_hooks.py(~38KB) — the server side:HookServer,ApiHookClient(the server-side helper class)src/api_hook_client.py(~31KB) — the client side:ApiHookClient(the Python client wrapper)
Architecture
┌─────────────────────────────────────────────────┐
│ Manual Slop (sloppy.py) │
│ - api_hooks.HookServer started in background │
│ - Listens on http://127.0.0.1:8999 │
└─────────────────┬───────────────────────────────┘
│ serves
▼
┌─────────────────────────────────────────────────┐
│ HTTP Endpoints │
│ GET /status - health │
│ GET /api/mma_status - MMA │
│ GET /api/gui - GUI │
│ POST /api/gui - push task │
│ POST /api/ask - blocking │
│ (synchronous HITL via HTTP) │
│ ... etc │
└─────────────────┬───────────────────────────────┘
│ consumed by
▼
┌─────────────────────────────────────────────────┐
│ src/api_hook_client.py: ApiHookClient │
│ Python wrapper. Used by: │
│ - live_gui test fixture (251 test files) │
│ - External scripts and agents │
│ - MMA conftest tooling │
└─────────────────────────────────────────────────┘
The /api/ask endpoint is special — it implements the Remote Confirmation Protocol for HITL approvals from external contexts. When the AI wants to run a script, the GUI can call /api/ask on a remote agent to get a yes/no answer.
The Server: HookServer (in src/api_hooks.py)
Lifecycle
The server is started by app_controller.py (or sloppy.py when --enable-test-hooks is passed):
# In app_controller.py or sloppy.py
from src.api_hooks import HookServer
server = HookServer()
server.start() # Background thread, non-blocking
The server runs in a daemon thread. It stops when the process exits (or via server.stop()).
Endpoints
| Method | Path | Purpose |
|---|---|---|
GET |
/status |
Health check. Returns JSON with status, pid, version, etc. |
GET |
/api/mma_status |
Full MMA orchestration state (tickets, progress, ETA) |
GET |
/api/gui |
GUI debug info (button IDs, fields, etc.) |
POST |
/api/gui |
Push a GUI task (click, set_value, custom_callback, etc.) |
POST |
/api/ask |
Synchronous HITL approval (blocking request/response) |
GET |
/api/performance |
Performance metrics (FPS, frame time) |
GET |
/api/comms |
Communication log |
GET |
/api/diagnostics |
Diagnostics state |
(Full endpoint list may grow; check the live server for the canonical list.)
Request/Response Format
All endpoints accept and return JSON:
// POST /api/gui
{
"action": "click",
"item": "btn_reset"
}
// Response
{
"status": "queued"
}
For blocking endpoints (/api/ask):
// POST /api/ask
{
"prompt": "Allow the AI to run 'rm -rf /'?",
"timeout": 30
}
// Response (blocks until human responds)
{
"approved": true,
"response": "yes go ahead"
}
The predefined_callbacks Mechanism
The server supports invoking any App method via the custom_callback action:
# Server side (in gui_2.py __init__)
self.controller._predefined_callbacks['_my_method'] = self._my_method
self.controller._gettable_fields['show_my_thing'] = 'show_my_thing'
# Client side
client.push_event("custom_callback", {
"callback": "_my_method",
"args": [arg1, arg2],
})
This is the standard way to expose new App methods to the Hook API.
The Client: ApiHookClient (in src/api_hook_client.py)
Connection
from src.api_hook_client import ApiHookClient
client = ApiHookClient() # connects to http://127.0.0.1:8999 by default
Custom host/port:
client = ApiHookClient(host="192.168.1.100", port=8999)
Key Methods
click(item, user_data=None)
Simulates a button click. The button must be registered as a clickable widget in the GUI.
client.click("btn_reset")
client.click("btn_save_project", user_data={"path": "/path/to/file"})
set_value(item, value)
Sets a widget's value (e.g., a text input).
client.set_value("ui_ai_input", "Hello world")
client.set_value("provider_select", "anthropic")
select_tab(item, value)
Selects a tab in a tab bar.
client.select_tab("main_tabs", "AI Settings")
select_list_item(item, value)
Selects an item in a listbox or combo.
client.select_list_item("model_select", "gemini-3-flash-preview")
push_event(action, payload)
Generic GUI task push. The action is dispatched to the appropriate handler.
client.push_event("custom_callback", {
"callback": "_my_method",
"args": [arg1, arg2],
})
get_value(item)
Reads a gettable field's current value.
value = client.get_value("show_command_palette") # Returns bool
value = client.get_value("current_provider") # Returns str
get_status()
Calls /status. Returns the parsed JSON.
status = client.get_status()
assert status["status"] == "running"
reset_session()
Calls the reset endpoint. Clears the GUI state for a fresh test.
client.reset_session()
wait_for_event(event_type, timeout=10)
Polls the event queue for a specific event type. Useful for waiting for async AI responses.
event = client.wait_for_event("ai_response", timeout=30)
assert event is not None
Robustness
The client has:
- Retry logic for transient connection failures
- Health check polling — auto-waits for the server to be ready
- Timeout configuration per call
The /api/ask Protocol (Remote Confirmation)
The /api/ask endpoint implements non-blocking, ID-based challenge/response for HITL approvals from external contexts.
Use Case
A headless service (running in a container, e.g., on Unraid) needs to ask the human "should the AI run this command?" The human is monitoring via a web client. The flow:
1. Headless service: POST /api/ask with prompt
2. Server: creates a challenge with unique ID
3. Server: pushes challenge to GUI's pending-ask queue
4. GUI: renders ask modal (or queues for later)
5. Human: clicks Approve or Reject in GUI
6. Server: unblocks the /api/ask call with the answer
7. Headless service: receives the response
This decouples the headless service from the GUI's render loop. The headless service doesn't need a direct connection to the GUI process.
Connection to the GUI
The App class has a few integration points with the Hook API:
In gui_2.py __init__
# Register predefined callbacks
self.controller._predefined_callbacks['save_context_preset'] = self.save_context_preset
self.controller._predefined_callbacks['load_context_preset'] = self.load_context_preset
self.controller._predefined_callbacks['_toggle_command_palette'] = self._toggle_command_palette
# ... etc
# Register gettable fields
self.controller._gettable_fields['show_command_palette'] = 'show_command_palette'
In app_controller.py
AppController._init_ai_and_hooks() starts the HookServer if --enable-test-hooks is set.
In sloppy.py
Parses --enable-test-hooks from CLI args and passes it to the controller.
Common Patterns
Test Pattern (from live_gui tests)
def test_my_feature(live_gui):
client = ApiHookClient()
assert client.wait_for_server(timeout=10)
# Open the command palette
client.push_event("custom_callback", {
"callback": "_toggle_command_palette",
"args": [],
})
time.sleep(0.5)
# Verify state
assert client.get_value("show_command_palette") is True
# Close
client.push_event("custom_callback", {
"callback": "_toggle_command_palette",
"args": [],
})
External Script Pattern
import requests
import time
def wait_for_app(url="http://127.0.0.1:8999", timeout=30):
start = time.time()
while time.time() - start < timeout:
try:
r = requests.get(f"{url}/status", timeout=1)
if r.status_code == 200:
return r.json()
except requests.ConnectionError:
pass
time.sleep(0.5)
raise RuntimeError("App did not start in time")
def trigger_hot_reload(url="http://127.0.0.1:8999"):
r = requests.post(f"{url}/api/gui", json={"action": "click", "item": "btn_hot_reload"})
r.raise_for_status()
return r.json()
Synchronous Ask Pattern
def ask_human_approval(prompt, url="http://127.0.0.1:8999", timeout=60):
r = requests.post(
f"{url}/api/ask",
json={"prompt": prompt, "timeout": timeout},
timeout=timeout + 5,
)
r.raise_for_status()
data = r.json()
return data.get("approved", False), data.get("response", "")
Testing
Unit Tests (tests/test_api_hook_client.py)
Tests the ApiHookClient class methods in isolation (mocked HTTP).
Live GUI Tests
Use the live_gui fixture which auto-launches sloppy.py --enable-test-hooks. The ApiHookClient then connects to the running app.
def test_via_hooks(live_gui):
client = ApiHookClient()
# ... use the client ...
Headless Service Tests
tests/test_headless_service.py and tests/test_headless_verification.py test the full server lifecycle without launching the GUI.
Limitations
- Single GUI instance per port: The server binds to a specific port. Multiple instances on the same port fail.
- No authentication: The Hook API has no auth. It's intended for localhost use. For remote access, use a reverse proxy with auth.
- Synchronous
/api/askblocks the server thread: Long timeouts can starve other requests. Use reasonable timeouts. - JSON-only payloads: All endpoints take and return JSON. Binary data is not supported.
See Also
- guide_architecture.md — Threading model
- guide_testing.md — How tests use this API
- guide_meta_boundary.md — Application vs Meta-Tooling domain separation
tests/conftest.py:live_gui— The session fixture that uses this APItests/test_api_hook_client.py— Client unit teststests/test_hooks.py— Basic HookServer smoke tests