state-hub/mcp_server/codex_server.py

"""Small State Hub MCP surface for Codex repository sessions.

The full dev-hub MCP server intentionally exposes administrative and catalog
operations. Codex repository work needs a much smaller coordination surface;
keeping it separate reduces tool discovery cost and makes the contract clear.
"""

from __future__ import annotations

import json
import os
from typing import Any

import httpx
from fastmcp import FastMCP

from mcp_server.server import get_domain_summary as _full_get_domain_summary


API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000").rstrip("/")

mcp = FastMCP(
    name="dev-hub-codex",
    instructions=(
        "Slim State Hub coordination surface for Codex repository sessions. "
        "Start with get_domain_summary, check the repository inbox, and record "
        "progress when work closes. Workplan files remain the source of truth."
    ),
)


def _request(method: str, path: str, body: dict[str, Any] | None = None) -> Any:
    with httpx.Client(
        base_url=API_BASE,
        timeout=15.0,
        follow_redirects=True,
        trust_env=False,
    ) as client:
        response = client.request(method, path, json=body)
        response.raise_for_status()
        return response.json()


def _json(value: Any) -> str:
    return json.dumps(value, indent=2)


@mcp.tool()
def get_domain_summary(domain_slug: str) -> str:
    """Return actionable State Hub orientation scoped to one domain."""
    return _full_get_domain_summary(domain_slug)


@mcp.tool()
def get_messages(to_agent: str, unread_only: bool = True) -> str:
    """Get coordination messages for one repository agent."""
    suffix = "true" if unread_only else "false"
    return _json(_request("GET", f"/messages/?to_agent={to_agent}&unread_only={suffix}"))


@mcp.tool()
def mark_message_read(message_id: str) -> str:
    """Mark one coordination message as read."""
    return _json(_request("PATCH", f"/messages/{message_id}/read", {}))


@mcp.tool()
def add_progress_event(
    summary: str,
    author: str = "codex",
    workplan_id: str | None = None,
    task_id: str | None = None,
) -> str:
    """Record a State Hub progress note for completed or significant work."""
    body: dict[str, Any] = {
        "summary": summary,
        "event_type": "note",
        "author": author,
    }
    if workplan_id:
        body["workplan_id"] = workplan_id
    if task_id:
        body["task_id"] = task_id
    return _json(_request("POST", "/progress/", body))


@mcp.tool()
def record_decision(
    title: str,
    description: str,
    topic_id: str,
    proposed_by: str = "codex",
    workplan_id: str | None = None,
) -> str:
    """Record a pending decision linked to a topic and optionally a workplan."""
    body: dict[str, Any] = {
        "title": title,
        "description": description,
        "topic_id": topic_id,
        "decision_type": "pending",
        "status": "open",
        "proposed_by": proposed_by,
    }
    if workplan_id:
        body["workplan_id"] = workplan_id
    return _json(_request("POST", "/decisions/", body))


@mcp.tool()
def update_task_status(
    task_id: str,
    status: str,
    blocking_reason: str | None = None,
) -> str:
    """Update a task to wait, todo, progress, done, or cancel."""
    body: dict[str, Any] = {"status": status}
    if blocking_reason is not None:
        body["blocking_reason"] = blocking_reason
    return _json(_request("PATCH", f"/tasks/{task_id}", body))


if __name__ == "__main__":
    mcp.run(transport="stdio")