PR Review MCP Server

"""MCP tools for GitHub PR review management.""" import json from typing import Any from mcp.server import Server from mcp.types import TextContent, Tool from .gh_api import GitHubAPI, GitHubAPIError def register_tools(server: Server) -> None: """Register all MCP tools with the server.""" api = GitHubAPI() # Tool 1: List review threads @server.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="list_review_threads", description="List review threads for a GitHub pull request", inputSchema={ "type": "object", "properties": { "owner": { "type": "string", "description": "Repository owner (username or organization)", }, "repo": {"type": "string", "description": "Repository name"}, "pull_number": {"type": "integer", "description": "Pull request number"}, "unresolved_only": { "type": "boolean", "description": "Only return unresolved threads (default: true)", "default": True, }, }, "required": ["owner", "repo", "pull_number"], }, ), Tool( name="reply_to_review_thread", description="Add a reply to a review thread", inputSchema={ "type": "object", "properties": { "owner": { "type": "string", "description": "Repository owner (username or organization)", }, "repo": {"type": "string", "description": "Repository name"}, "pull_number": {"type": "integer", "description": "Pull request number"}, "thread_id": { "type": "string", "description": "Review thread ID (from list_review_threads)", }, "body": { "type": "string", "description": "Reply content (Markdown supported)", }, }, "required": ["owner", "repo", "pull_number", "thread_id", "body"], }, ), Tool( name="resolve_review_thread", description="Mark a review thread as resolved", inputSchema={ "type": "object", "properties": { "thread_id": { "type": "string", "description": "Review thread ID (from list_review_threads)", } }, "required": ["thread_id"], }, ), Tool( name="reply_and_resolve", description="Reply to a review thread and immediately resolve it", inputSchema={ "type": "object", "properties": { "owner": { "type": "string", "description": "Repository owner (username or organization)", }, "repo": {"type": "string", "description": "Repository name"}, "pull_number": {"type": "integer", "description": "Pull request number"}, "thread_id": { "type": "string", "description": "Review thread ID (from list_review_threads)", }, "body": { "type": "string", "description": "Reply content (Markdown supported)", }, }, "required": ["owner", "repo", "pull_number", "thread_id", "body"], }, ), ] @server.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: """Handle tool calls.""" try: if name == "list_review_threads": return await handle_list_review_threads(api, arguments) elif name == "reply_to_review_thread": return await handle_reply_to_review_thread(api, arguments) elif name == "resolve_review_thread": return await handle_resolve_review_thread(api, arguments) elif name == "reply_and_resolve": return await handle_reply_and_resolve(api, arguments) else: raise ValueError(f"Unknown tool: {name}") except GitHubAPIError as e: return [TextContent(type="text", text=f"Error: {str(e)}")] except Exception as e: return [TextContent(type="text", text=f"Unexpected error: {str(e)}")] async def handle_list_review_threads( api: GitHubAPI, arguments: dict[str, Any] ) -> list[TextContent]: """Handle list_review_threads tool call.""" owner = arguments["owner"] repo = arguments["repo"] pull_number = arguments["pull_number"] unresolved_only = arguments.get("unresolved_only", True) threads = api.list_review_threads(owner, repo, pull_number, unresolved_only) # Format output result = { "pull_request": f"{owner}/{repo}#{pull_number}", "thread_count": len(threads), "threads": [], } for thread in threads: comments = thread.get("comments", {}).get("nodes", []) first_comment = comments[0] if comments else {} thread_info = { "id": thread.get("id"), "is_resolved": thread.get("isResolved", False), "file": thread.get("path"), "line": thread.get("line"), "start_line": thread.get("startLine"), "diff_side": thread.get("diffSide"), "comment_count": len(comments), "first_comment": { "author": first_comment.get("author", {}).get("login", "unknown"), "body": first_comment.get("body", ""), "created_at": first_comment.get("createdAt", ""), }, } result["threads"].append(thread_info) return [TextContent(type="text", text=json.dumps(result, indent=2))] async def handle_reply_to_review_thread( api: GitHubAPI, arguments: dict[str, Any] ) -> list[TextContent]: """Handle reply_to_review_thread tool call.""" owner = arguments["owner"] repo = arguments["repo"] pull_number = arguments["pull_number"] thread_id = arguments["thread_id"] body = arguments["body"] # Get PR ID pr_id = api.get_pr_id(owner, repo, pull_number) # Add reply comment = api.add_thread_reply(pr_id, thread_id, body) result = { "success": True, "comment": { "id": comment.get("id"), "author": comment.get("author", {}).get("login"), "body": comment.get("body"), "created_at": comment.get("createdAt"), }, } return [TextContent(type="text", text=json.dumps(result, indent=2))] async def handle_resolve_review_thread( api: GitHubAPI, arguments: dict[str, Any] ) -> list[TextContent]: """Handle resolve_review_thread tool call.""" thread_id = arguments["thread_id"] # Resolve thread thread = api.resolve_thread(thread_id) result = { "success": True, "thread": {"id": thread.get("id"), "is_resolved": thread.get("isResolved", False)}, } return [TextContent(type="text", text=json.dumps(result, indent=2))] async def handle_reply_and_resolve(api: GitHubAPI, arguments: dict[str, Any]) -> list[TextContent]: """Handle reply_and_resolve tool call.""" owner = arguments["owner"] repo = arguments["repo"] pull_number = arguments["pull_number"] thread_id = arguments["thread_id"] body = arguments["body"] # Get PR ID pr_id = api.get_pr_id(owner, repo, pull_number) # Add reply comment = api.add_thread_reply(pr_id, thread_id, body) # Resolve thread thread = api.resolve_thread(thread_id) result = { "success": True, "comment": { "id": comment.get("id"), "author": comment.get("author", {}).get("login"), "body": comment.get("body"), "created_at": comment.get("createdAt"), }, "thread": {"id": thread.get("id"), "is_resolved": thread.get("isResolved", False)}, } return [TextContent(type="text", text=json.dumps(result, indent=2))]