Private
Public Access
0
0
Files
manual_slop/docs/guide_api_hooks.md
T

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/gui/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/gui/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

  1. Single GUI instance per port: The server binds to a specific port. Multiple instances on the same port fail.
  2. No authentication: The Hook API has no auth. It's intended for localhost use. For remote access, use a reverse proxy with auth.
  3. Synchronous /api/ask blocks the server thread: Long timeouts can starve other requests. Use reasonable timeouts.
  4. 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 API
  • tests/test_api_hook_client.py — Client unit tests
  • tests/test_hooks.py — Basic HookServer smoke tests