feat: Add Gitea MCP client for autonomous PR workflow

src/timmy/gitea_mcp_client.py

@@ -0,0 +1,227 @@
+"""Gitea MCP Client for Timmy.
+
+Connects to the Gitea MCP Server and exposes repository management,
+PR creation, and code review tools to the agent.
+"""
+
+import json
+import logging
+import os
+import subprocess
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+class GiteaMCPClient:
+    """Client for interacting with Gitea via MCP protocol."""
+
+    def __init__(
+        self,
+        gitea_host: str | None = None,
+        gitea_token: str | None = None,
+        mcp_server_path: str = "gitea-mcp",
+    ):
+        """Initialize the Gitea MCP client.
+
+        Args:
+            gitea_host: Gitea instance URL (e.g., http://100.124.176.28:3000)
+            gitea_token: Personal access token for Gitea
+            mcp_server_path: Path to gitea-mcp binary
+        """
+        self.gitea_host = gitea_host or os.getenv("GITEA_HOST", "http://localhost:3000")
+        self.gitea_token = gitea_token or os.getenv("GITEA_ACCESS_TOKEN", "")
+        self.mcp_server_path = mcp_server_path
+        self.process = None
+
+    def start(self) -> bool:
+        """Start the MCP server in stdio mode.
+
+        Returns:
+            True if server started successfully.
+        """
+        if self.process is not None:
+            logger.warning("MCP server already running")
+            return True
+
+        try:
+            env = os.environ.copy()
+            env["GITEA_HOST"] = self.gitea_host
+            env["GITEA_ACCESS_TOKEN"] = self.gitea_token
+
+            self.process = subprocess.Popen(
+                [self.mcp_server_path, "--mode", "stdio"],
+                stdin=subprocess.PIPE,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                env=env,
+                text=True,
+            )
+            logger.info("Gitea MCP server started")
+            return True
+        except FileNotFoundError:
+            logger.error(f"MCP server binary not found: {self.mcp_server_path}")
+            return False
+        except Exception as e:
+            logger.error(f"Failed to start MCP server: {e}")
+            return False
+
+    def stop(self) -> None:
+        """Stop the MCP server."""
+        if self.process:
+            self.process.terminate()
+            try:
+                self.process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+            self.process = None
+            logger.info("Gitea MCP server stopped")
+
+    def call_tool(self, tool_name: str, **kwargs) -> dict[str, Any]:
+        """Call a Gitea MCP tool.
+
+        Args:
+            tool_name: Name of the tool (e.g., "list_my_repos")
+            **kwargs: Tool arguments
+
+        Returns:
+            Tool result as a dictionary.
+        """
+        if not self.process:
+            self.start()
+
+        request = {
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "tools/call",
+            "params": {
+                "name": tool_name,
+                "arguments": kwargs,
+            },
+        }
+
+        try:
+            self.process.stdin.write(json.dumps(request) + "\n")
+            self.process.stdin.flush()
+
+            response_line = self.process.stdout.readline()
+            response = json.loads(response_line)
+
+            if "result" in response:
+                return response["result"]
+            elif "error" in response:
+                logger.error(f"Tool error: {response['error']}")
+                return {"error": response["error"]}
+            else:
+                return response
+        except Exception as e:
+            logger.error(f"Failed to call tool {tool_name}: {e}")
+            return {"error": str(e)}
+
+    def list_repos(self) -> list[dict]:
+        """List all repositories owned by the authenticated user."""
+        result = self.call_tool("list_my_repos")
+        return result.get("repositories", []) if isinstance(result, dict) else []
+
+    def get_repo(self, owner: str, repo: str) -> dict:
+        """Get repository details."""
+        repos = self.list_repos()
+        for r in repos:
+            if r.get("owner") == owner and r.get("name") == repo:
+                return r
+        return {}
+
+    def create_pull_request(
+        self,
+        owner: str,
+        repo: str,
+        title: str,
+        body: str,
+        head: str,
+        base: str = "main",
+    ) -> dict:
+        """Create a new pull request."""
+        return self.call_tool(
+            "create_pull_request",
+            owner=owner,
+            repo=repo,
+            title=title,
+            body=body,
+            head=head,
+            base=base,
+        )
+
+    def list_pull_requests(self, owner: str, repo: str) -> list[dict]:
+        """List all pull requests in a repository."""
+        result = self.call_tool(
+            "list_repo_pull_requests",
+            owner=owner,
+            repo=repo,
+        )
+        return result.get("pull_requests", []) if isinstance(result, dict) else []
+
+    def create_issue(
+        self,
+        owner: str,
+        repo: str,
+        title: str,
+        body: str = "",
+    ) -> dict:
+        """Create a new issue."""
+        return self.call_tool(
+            "create_issue",
+            owner=owner,
+            repo=repo,
+            title=title,
+            body=body,
+        )
+
+    def get_file_content(self, owner: str, repo: str, path: str) -> dict:
+        """Get the content of a file."""
+        return self.call_tool(
+            "get_file_content",
+            owner=owner,
+            repo=repo,
+            path=path,
+        )
+
+    def create_file(
+        self,
+        owner: str,
+        repo: str,
+        path: str,
+        content: str,
+        message: str,
+        branch: str = "main",
+    ) -> dict:
+        """Create a new file."""
+        return self.call_tool(
+            "create_file",
+            owner=owner,
+            repo=repo,
+            path=path,
+            content=content,
+            message=message,
+            branch=branch,
+        )
+
+    def update_file(
+        self,
+        owner: str,
+        repo: str,
+        path: str,
+        content: str,
+        message: str,
+        branch: str = "main",
+    ) -> dict:
+        """Update an existing file."""
+        return self.call_tool(
+            "update_file",
+            owner=owner,
+            repo=repo,
+            path=path,
+            content=content,
+            message=message,
+            branch=branch,
+        )