agentrunner/mcp_launcher.py

#!/usr/bin/env python3
"""
MCP Server for ClearGrow orchestrator.

Communicates with the agent runner via Unix socket to provide
YouTrack and agent pool data to the orchestrator.
"""

import json
import socket
import sys

SOCKET_PATH = "/run/cleargrow/runner.sock"


def call_api(method: str, params: dict = None) -> dict:
    """Call the internal API via Unix socket."""
    try:
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.settimeout(10.0)
        sock.connect(SOCKET_PATH)

        request = json.dumps({"method": method, "params": params or {}})
        sock.sendall(request.encode() + b"\n")

        response = b""
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
            if b"\n" in response:
                break

        sock.close()
        return json.loads(response.decode().strip())

    except socket.timeout:
        return {"error": "Request timed out"}
    except FileNotFoundError:
        return {"error": "Agent runner not available (socket not found)"}
    except Exception as e:
        return {"error": str(e)}


TOOLS = [
    {
        "name": "get_issues",
        "description": "Get issues from YouTrack by workflow state. Returns ID, summary, type, priority, and description snippet for each issue.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "state": {
                    "type": "string",
                    "description": "Workflow state: Backlog, Triage, Ready, 'In Progress', Build, Verify, Document, or Review",
                },
                "limit": {
                    "type": "integer",
                    "description": "Max issues to return (default 20)",
                    "default": 20,
                },
            },
            "required": ["state"],
        },
    },
    {
        "name": "get_issue_details",
        "description": "Get full details of a specific issue including complete description.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "issue_id": {
                    "type": "string",
                    "description": "Issue ID (e.g., CG-123)",
                },
            },
            "required": ["issue_id"],
        },
    },
    {
        "name": "get_agent_pool_status",
        "description": "Get current agent pool status - how many agents are active and what tasks they're running.",
        "inputSchema": {
            "type": "object",
            "properties": {},
        },
    },
    {
        "name": "get_workflow_summary",
        "description": "Get a summary of all issues across all workflow states. Good starting point for analysis.",
        "inputSchema": {
            "type": "object",
            "properties": {},
        },
    },
    {
        "name": "transition_issue",
        "description": "Move an issue to a different workflow state.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "issue_id": {
                    "type": "string",
                    "description": "Issue ID to transition (e.g., CG-123)",
                },
                "new_state": {
                    "type": "string",
                    "description": "Target state: Ready, 'In Progress', Build, Verify, Document, Review, or Done",
                },
            },
            "required": ["issue_id", "new_state"],
        },
    },
    {
        "name": "add_comment",
        "description": "Add a comment to an issue.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "issue_id": {
                    "type": "string",
                    "description": "Issue ID (e.g., CG-123)",
                },
                "text": {
                    "type": "string",
                    "description": "Comment text to add",
                },
            },
            "required": ["issue_id", "text"],
        },
    },
]


def handle_tool_call(name: str, arguments: dict) -> str:
    """Execute a tool call and return the result as formatted text."""
    result = call_api(name, arguments)
    return json.dumps(result, indent=2)


def main():
    """Run MCP server on stdio."""
    while True:
        try:
            line = sys.stdin.readline()
            if not line:
                break

            request = json.loads(line.strip())
            method = request.get("method", "")
            params = request.get("params", {})
            request_id = request.get("id")

            response = None

            if method == "initialize":
                response = {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "result": {
                        "protocolVersion": "2024-11-05",
                        "capabilities": {"tools": {}},
                        "serverInfo": {"name": "cleargrow", "version": "1.0.0"},
                    },
                }

            elif method == "tools/list":
                response = {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "result": {"tools": TOOLS},
                }

            elif method == "tools/call":
                tool_name = params.get("name", "")
                tool_args = params.get("arguments", {})
                result_text = handle_tool_call(tool_name, tool_args)
                response = {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "result": {"content": [{"type": "text", "text": result_text}]},
                }

            elif method == "notifications/initialized":
                pass  # No response for notifications

            else:
                response = {
                    "jsonrpc": "2.0",
                    "id": request_id,
                    "error": {"code": -32601, "message": f"Unknown method: {method}"},
                }

            if response:
                sys.stdout.write(json.dumps(response) + "\n")
                sys.stdout.flush()

        except json.JSONDecodeError:
            continue
        except Exception as e:
            sys.stderr.write(f"MCP Error: {e}\n")
            sys.stderr.flush()


if __name__ == "__main__":
    main()