joy-cli/bin/joy-mcp-server.py

#!/usr/bin/env python3
"""joy-mcp-server — 将 joy CLI 高频子命令暴露为 MCP tools.

Phase 1（tux 决策 2026-07-05）：包 5 个高频子命令，让 konqi/tux 用 MCP tool call
替代 bash 拼命令。神兽保持调 joy CLI 不变。

Transport: stdio（随 agent 生灭，零生命周期管理）。
依赖: fastmcp（venv ~/Projects/joy-cli/.venv）。

Tools:
  - tmux_check      → joy tmux --check      （可观测性快照）
  - clone_status    → joy clone status      （神兽 pi 进程状态）
  - work_transcript → joy work <vm> transcript [N]  （读 work session 屏幕）
  - work_inject     → joy work <vm> inject "msg"     （向 work session 注入指令）
  - board_create    → joy board create <vm> "desc"   （创建 board task）
"""
from __future__ import annotations

import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any

from fastmcp import FastMCP

# ── joy 二进制定位 ──────────────────────────────────────────────
# MCP server 可能不继承完整 shell env，优先用绝对路径，回退 PATH 查找。
_JOY_BIN = str(Path(__file__).resolve().parent / "joy")
if not os.path.isfile(_JOY_BIN) or not os.access(_JOY_BIN, os.X_OK):
    _JOY_BIN = shutil.which("joy") or "joy"

# joy CLI 依赖的环境变量（与 shell 中一致）
_LINUXJOY = os.environ.get("LINUXJOY", str(Path.home() / "Projects" / "linuxjoy"))


def _run_joy(args: list[str], timeout: int = 30) -> tuple[int, str, str]:
    """执行 joy 子命令，返回 (returncode, stdout, stderr)。"""
    env = dict(os.environ)
    env["LINUXJOY"] = _LINUXJOY
    try:
        proc = subprocess.run(
            [_JOY_BIN] + args,
            capture_output=True,
            text=True,
            timeout=timeout,
            env=env,
        )
        return proc.returncode, proc.stdout, proc.stderr
    except subprocess.TimeoutExpired:
        return 124, "", f"joy {' '.join(args)} timed out after {timeout}s"
    except FileNotFoundError:
        return 127, "", f"joy binary not found: {_JOY_BIN}"


def _parse_json(stdout: str) -> Any:
    """尝试解析 JSON，失败则返回原始字符串。"""
    try:
        return json.loads(stdout)
    except (json.JSONDecodeError, ValueError):
        return stdout.strip()


def _result(rc: int, stdout: str, stderr: str, parse_json: bool = True) -> dict[str, Any]:
    """统一结果封装：成功返回 data，失败返回 error。"""
    if rc != 0:
        return {
            "ok": False,
            "error": stderr.strip() or stdout.strip() or f"joy exited {rc}",
            "returncode": rc,
        }
    return {
        "ok": True,
        "data": _parse_json(stdout) if parse_json else stdout.strip(),
    }


# ── MCP Server ─────────────────────────────────────────────────
mcp = FastMCP("joy", instructions="linuxjoy joy CLI 工具集（巡检/派活/board）。")


@mcp.tool()
def tmux_check() -> dict[str, Any]:
    """可观测性快照：tmux 会话存在性 + pi 进程数 + 异常列表。

    等价 `joy tmux --check`。每轮巡检第一步，快速判断集群健康。
    返回 JSON: {joy-clones, clone-panes, joy-works, works-panes,
               joy-konqi, joy-tux, pi_count, issues[]}.
    """
    rc, out, err = _run_joy(["tmux", "--check"])
    return _result(rc, out, err)


@mcp.tool()
def clone_status() -> dict[str, Any]:
    """神兽 pi 进程状态：每个 clone 的 title + pid。

    等价 `joy clone status`。判断哪些神兽活着、哪些在 sleep/bash。
    返回 JSON: {running, count, panes[{title, pid}]}.
    """
    rc, out, err = _run_joy(["clone", "status"])
    return _result(rc, out, err)


@mcp.tool()
def work_transcript(vm: str, lines: int = 20) -> dict[str, Any]:
    """读取指定 VM work session 的屏幕输出（最近 N 行）。

    等价 `joy work <vm> transcript [N]`。用于判断 work 在干什么、
    是否卡住、是否 idle。lines 默认 20，快速检查用 5-10 行。

    Args:
        vm: VM 名称（如 devops, wenpai, weixiaoduo）
        lines: 读取行数，默认 20
    """
    rc, out, err = _run_joy(["work", vm, "transcript", str(lines)])
    return _result(rc, out, err, parse_json=False)


@mcp.tool()
def work_inject(vm: str, message: str) -> dict[str, Any]:
    """向指定 VM 的 work session 注入指令（写入输入缓冲）。

    等价 `joy work <vm> inject "指令"`。用于派活、催办、纠偏。
    注意：这是写操作，会直接影响 VM 上的 work session。

    Args:
        vm: 目标 VM 名称
        message: 要注入的指令文本
    """
    rc, out, err = _run_joy(["work", vm, "inject", message], timeout=15)
    return _result(rc, out, err, parse_json=False)


@mcp.tool()
def board_create(vm: str, description: str) -> dict[str, Any]:
    """创建 board task 并分配给指定 VM。

    等价 `joy board create <vm> "描述"`。board 是 linuxjoy 本体唯一语言，
    重要发现/修复/决策都应创建 task。

    Args:
        vm: 目标 VM 名称（task target）
        description: 任务描述
    """
    rc, out, err = _run_joy(["board", "create", vm, description])
    return _result(rc, out, err)


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