agentrunner/mcp_server.py

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

Exposes internal APIs as tools for the orchestrator agent.
"""

import json
import sys
from typing import Any

# MCP protocol constants
JSONRPC_VERSION = "2.0"


class MCPServer:
    """Simple MCP server exposing agent runner APIs."""

    def __init__(self, runner):
        self.runner = runner
        self.tools = {
            "get_issues": self.get_issues,
            "get_issue": self.get_issue,
            "get_agent_status": self.get_agent_status,
            "transition_issue": self.transition_issue,
            "add_comment": self.add_comment,
        }

    def get_tool_definitions(self) -> list[dict]:
        """Return tool definitions for MCP."""
        return [
            {
                "name": "get_issues",
                "description": "Get issues from YouTrack by state. Returns issue ID, summary, type, priority, description, and all comments for each issue.",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "state": {
                            "type": "string",
                            "description": "Issue state to filter by (Triage, Ready, In Progress, Build, Verify, Document, Review)",
                        },
                        "limit": {
                            "type": "integer",
                            "description": "Maximum number of issues to return (default 20)",
                            "default": 20,
                        },
                    },
                    "required": ["state"],
                },
            },
            {
                "name": "get_issue",
                "description": "Get detailed information about a specific issue by ID, including all comments. Use this to understand issue history and current status.",
                "inputSchema": {
                    "type": "object",
                    "properties": {
                        "issue_id": {
                            "type": "string",
                            "description": "Issue ID (e.g., CG-123)",
                        },
                    },
                    "required": ["issue_id"],
                },
            },
            {
                "name": "get_agent_status",
                "description": "Get current status of the agent pool including active agents and running tasks.",
                "inputSchema": {
                    "type": "object",
                    "properties": {},
                },
            },
            {
                "name": "transition_issue",
                "description": "Move an issue to a new state. Use this to suggest state transitions.",
                "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, etc.)",
                        },
                    },
                    "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 to comment on (e.g., CG-123)",
                        },
                        "text": {
                            "type": "string",
                            "description": "Comment text",
                        },
                    },
                    "required": ["issue_id", "text"],
                },
            },
        ]

    def get_issues(self, state: str, limit: int = 20) -> dict:
        """Get issues by state, including comments for each issue."""
        try:
            project = self.runner.config.get("project", {}).get("name", "CG")
            issues = self.runner.youtrack.get_issues_by_state(project, state)

            if not issues:
                return {"issues": [], "count": 0, "state": state}

            result = []
            for issue in issues[:limit]:
                # Get comments for this issue
                comments = []
                try:
                    raw_comments = self.runner.youtrack.get_issue_comments(issue.id)
                    for c in raw_comments:
                        comments.append({
                            "author": getattr(c, "author", "Unknown"),
                            "text": getattr(c, "text", ""),
                            "created": str(getattr(c, "created", "")),
                        })
                except Exception:
                    pass

                result.append({
                    "id": issue.id,
                    "summary": issue.summary,
                    "type": getattr(issue, "type", "Task"),
                    "priority": getattr(issue, "priority", "Normal"),
                    "description": getattr(issue, "description", ""),
                    "created": str(getattr(issue, "created", "")),
                    "comments": comments,
                })

            return {"issues": result, "count": len(issues), "state": state}

        except Exception as e:
            return {"error": str(e)}

    def get_issue(self, issue_id: str) -> dict:
        """Get single issue details including comments."""
        try:
            issue = self.runner.youtrack.get_issue(issue_id)
            if not issue:
                return {"error": f"Issue {issue_id} not found"}

            # Get comments for this issue
            comments = []
            try:
                raw_comments = self.runner.youtrack.get_issue_comments(issue_id)
                for c in raw_comments:
                    comments.append({
                        "author": getattr(c, "author", "Unknown"),
                        "text": getattr(c, "text", ""),
                        "created": str(getattr(c, "created", "")),
                    })
            except Exception:
                # Don't fail if comments can't be fetched
                pass

            return {
                "id": issue.id,
                "summary": issue.summary,
                "type": getattr(issue, "type", "Task"),
                "priority": getattr(issue, "priority", "Normal"),
                "state": getattr(issue, "state", "Unknown"),
                "description": getattr(issue, "description", ""),
                "created": str(getattr(issue, "created", "")),
                "updated": str(getattr(issue, "updated", "")),
                "comments": comments,
            }

        except Exception as e:
            return {"error": str(e)}

    def get_agent_status(self) -> dict:
        """Get agent pool status."""
        try:
            status = self.runner.agent_pool.get_status()
            return {
                "active": status.get("active", 0),
                "max_agents": status.get("max_agents", 10),
                "tasks": status.get("tasks", []),
            }
        except Exception as e:
            return {"error": str(e)}

    def transition_issue(self, issue_id: str, new_state: str) -> dict:
        """Transition an issue to a new state."""
        try:
            self.runner.youtrack.update_issue_state(issue_id, new_state)
            return {"success": True, "issue_id": issue_id, "new_state": new_state}
        except Exception as e:
            return {"error": str(e)}

    def add_comment(self, issue_id: str, text: str) -> dict:
        """Add a comment to an issue."""
        try:
            self.runner.youtrack.add_comment(issue_id, text)
            return {"success": True, "issue_id": issue_id}
        except Exception as e:
            return {"error": str(e)}

    def handle_request(self, request: dict) -> dict:
        """Handle a JSON-RPC request."""
        method = request.get("method", "")
        params = request.get("params", {})
        request_id = request.get("id")

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

        elif method == "tools/list":
            return {
                "jsonrpc": JSONRPC_VERSION,
                "id": request_id,
                "result": {"tools": self.get_tool_definitions()},
            }

        elif method == "tools/call":
            tool_name = params.get("name", "")
            tool_args = params.get("arguments", {})

            if tool_name not in self.tools:
                return {
                    "jsonrpc": JSONRPC_VERSION,
                    "id": request_id,
                    "error": {"code": -32601, "message": f"Unknown tool: {tool_name}"},
                }

            try:
                result = self.tools[tool_name](**tool_args)
                return {
                    "jsonrpc": JSONRPC_VERSION,
                    "id": request_id,
                    "result": {
                        "content": [{"type": "text", "text": json.dumps(result, indent=2)}],
                    },
                }
            except Exception as e:
                return {
                    "jsonrpc": JSONRPC_VERSION,
                    "id": request_id,
                    "error": {"code": -32000, "message": str(e)},
                }

        elif method == "notifications/initialized":
            # Client notification, no response needed
            return None

        else:
            return {
                "jsonrpc": JSONRPC_VERSION,
                "id": request_id,
                "error": {"code": -32601, "message": f"Unknown method: {method}"},
            }

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

                request = json.loads(line)
                response = self.handle_request(request)

                if response:  # Don't send response for notifications
                    sys.stdout.write(json.dumps(response) + "\n")
                    sys.stdout.flush()

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


def create_mcp_config(socket_path: str = "/tmp/cleargrow-mcp.sock") -> dict:
    """Create MCP configuration for Claude."""
    return {
        "mcpServers": {
            "cleargrow": {
                "command": sys.executable,
                "args": [__file__, "--stdio"],
                "env": {},
            }
        }
    }


if __name__ == "__main__":
    # For testing - in production, runner will instantiate this
    print("MCP Server for ClearGrow Agent Runner")
    print("Run with --stdio for MCP protocol mode")