From e2e4cbfe1ddfe77697f5908536674d7134b6071d Mon Sep 17 00:00:00 2001 From: Manus Agent Date: Sat, 14 Mar 2026 10:17:58 -0400 Subject: [PATCH] feat: Add Gitea MCP client for autonomous PR workflow --- src/timmy/gitea_mcp_client.py | 227 ++++++++++++++++++++++++++++++++++ 1 file changed, 227 insertions(+) create mode 100644 src/timmy/gitea_mcp_client.py diff --git a/src/timmy/gitea_mcp_client.py b/src/timmy/gitea_mcp_client.py new file mode 100644 index 00000000..22d80fba --- /dev/null +++ b/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, + )