orchestrator/sea-haven-claude-plugin/hooks/user_prompt_submit.py

#!/usr/bin/env python3
"""UserPromptSubmit hook: auto-delegate prefixed prompts to the agent-team API.

Claude Code calls this script via the UserPromptSubmit hook whenever the user
submits a prompt. If the prompt starts with ``/delegate`` (or ``DELEGATE:``),
the hook forwards the task description to the agent-team HTTP API
(``POST /tasks``) and blocks the prompt — the task is now running in the
pipeline; no need to also answer it locally.

Prompts that don't match the delegate prefix are passed through unchanged
(exit 0 with no output).

Configuration (env vars):
  AGENT_TEAM_API_URL    Base URL of the agent-team HTTP API
                        (default: http://127.0.0.1:8765)
  AGENT_TEAM_API_TOKEN  Bearer token from ``run-team.py serve``
                        (required when the API has token auth enabled)

Claude Code hook contract:
  * stdin:  JSON object with at least a ``prompt`` key.
  * stdout: JSON object or empty.
    - Empty / exit 0 → pass through (Claude answers normally).
    - ``{"action": "block", "reason": "..."}`` → block the prompt; Claude
      shows the reason to the user instead of answering.
  * exit 0 = allowed (or delegated+blocked); non-zero = block with error.
"""

from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request
from typing import Any

_DEFAULT_API_URL = "http://127.0.0.1:8765"
_DELEGATE_PREFIXES = ("/delegate ", "DELEGATE: ")


def _should_delegate(prompt: str) -> tuple[bool, str]:
    """Return (True, task_text) when the prompt is a delegate command, else (False, '').

    Matches ``/delegate <text>``, ``/delegate`` (bare — no text → usage hint),
    and ``DELEGATE: <text>``.
    """
    stripped = prompt.strip()
    for prefix in _DELEGATE_PREFIXES:
        if stripped.startswith(prefix):
            return True, stripped[len(prefix) :].strip()
    # Bare /delegate with no trailing space or text.
    if stripped == "/delegate":
        return True, ""
    return False, ""


def _call_api(task: str, *, api_url: str, token: str) -> dict[str, Any]:
    """POST /tasks and return the parsed response dict.

    Raises urllib.error.URLError / urllib.error.HTTPError on network / HTTP
    errors; the caller turns these into a block reason so Claude shows the error
    to the user rather than silently passing through.
    """
    payload = json.dumps({"task": task, "transport": "claude_code"}).encode("utf-8")
    headers: dict[str, str] = {"Content-Type": "application/json"}
    if token:
        headers["Authorization"] = f"Bearer {token}"
    req = urllib.request.Request(
        f"{api_url.rstrip('/')}/tasks",
        data=payload,
        headers=headers,
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
        return json.loads(resp.read().decode("utf-8"))


def run(
    prompt: str, *, api_url: str = _DEFAULT_API_URL, token: str = ""
) -> dict[str, Any]:
    """Core hook logic — pure function, fully testable without stdin/stdout.

    Returns the hook output dict:
    - ``{}`` to pass through (no delegation);
    - ``{"action": "block", "reason": "..."}`` to block and show a message.
    """
    should, task_text = _should_delegate(prompt)
    if not should:
        return {}

    if not task_text:
        return {
            "action": "block",
            "reason": (
                "Usage: /delegate <task description>\n"
                "Example: /delegate Add OAuth2 to the admin portal"
            ),
        }

    if not token:
        env_token = os.environ.get("AGENT_TEAM_API_TOKEN", "")
        if not env_token:
            return {
                "action": "block",
                "reason": (
                    "AGENT_TEAM_API_TOKEN is not set. "
                    "Start the agent-team server with `run-team.py serve` "
                    "and export the token it prints."
                ),
            }
        token = env_token

    try:
        result = _call_api(task_text, api_url=api_url, token=token)
    except urllib.error.HTTPError as exc:
        return {
            "action": "block",
            "reason": f"agent-team API error {exc.code}: {exc.reason}. Is the server running?",
        }
    except urllib.error.URLError as exc:
        return {
            "action": "block",
            "reason": (
                f"Cannot reach agent-team API at {api_url}: {exc.reason}. "
                "Is `run-team.py serve` running?"
            ),
        }
    except Exception as exc:  # noqa: BLE001
        return {
            "action": "block",
            "reason": f"Unexpected error delegating task: {exc}",
        }

    thread_id = result.get("thread_id", "(unknown)")
    return {
        "action": "block",
        "reason": (
            f"Task delegated to agent-team pipeline.\n"
            f"thread_id: {thread_id}\n"
            f"The coordinator will send a clarifying question via Slack."
        ),
    }


def main() -> int:
    try:
        data = json.loads(sys.stdin.read())
    except (json.JSONDecodeError, OSError):
        return 0  # pass through on bad input

    prompt = data.get("prompt", "") if isinstance(data, dict) else ""
    api_url = os.environ.get("AGENT_TEAM_API_URL", _DEFAULT_API_URL)
    token = os.environ.get("AGENT_TEAM_API_TOKEN", "")

    output = run(prompt, api_url=api_url, token=token)
    if output:
        print(json.dumps(output))
    return 0


if __name__ == "__main__":
    sys.exit(main())