PR Review MCP Server

"""GitHub GraphQL API wrapper using gh CLI.""" import json import subprocess from typing import Any class GitHubAPIError(Exception): """Exception raised for GitHub API errors.""" pass class GitHubAPI: """GitHub GraphQL API wrapper using gh CLI.""" @staticmethod def execute_graphql(query: str, variables: dict[str, Any]) -> dict[str, Any]: """ Execute a GraphQL query using gh CLI. Args: query: GraphQL query string variables: Query variables Returns: GraphQL response data Raises: GitHubAPIError: If the query fails or returns errors """ try: # Build gh api graphql command cmd = ["gh", "api", "graphql"] cmd.extend(["-f", f"query={query}"]) # Add variables with proper type conversion for key, value in variables.items(): if isinstance(value, bool): # Boolean values use -F flag cmd.extend(["-F", f"{key}={str(value).lower()}"]) elif isinstance(value, int): # Integer values use -F flag cmd.extend(["-F", f"{key}={value}"]) else: # String values use -f flag cmd.extend(["-f", f"{key}={value}"]) # Execute command result = subprocess.run(cmd, capture_output=True, text=True, check=True) # Parse response data = json.loads(result.stdout) # Check for GraphQL errors if "errors" in data: error_messages = [err.get("message", str(err)) for err in data["errors"]] raise GitHubAPIError(f"GraphQL errors: {', '.join(error_messages)}") return data.get("data", {}) except subprocess.CalledProcessError as e: raise GitHubAPIError(f"gh command failed: {e.stderr}") from e except json.JSONDecodeError as e: raise GitHubAPIError(f"Invalid JSON response: {e}") from e def get_pr_id(self, owner: str, repo: str, pull_number: int) -> str: """ Get the node ID of a pull request. Args: owner: Repository owner repo: Repository name pull_number: Pull request number Returns: Pull request node ID """ query = """ query GetPRId($owner: String!, $repo: String!, $pullNumber: Int!) { repository(owner: $owner, name: $repo) { pullRequest(number: $pullNumber) { id } } } """ variables = {"owner": owner, "repo": repo, "pullNumber": pull_number} data = self.execute_graphql(query, variables) pr_id = data.get("repository", {}).get("pullRequest", {}).get("id") if not pr_id: raise GitHubAPIError(f"Pull request #{pull_number} not found") return pr_id def list_review_threads( self, owner: str, repo: str, pull_number: int, unresolved_only: bool = True ) -> list[dict[str, Any]]: """ List review threads for a pull request. Args: owner: Repository owner repo: Repository name pull_number: Pull request number unresolved_only: Only return unresolved threads Returns: List of review thread objects """ query = """ query GetReviewThreads($owner: String!, $repo: String!, $pullNumber: Int!) { repository(owner: $owner, name: $repo) { pullRequest(number: $pullNumber) { reviewThreads(first: 100) { nodes { id isResolved path line startLine diffSide comments(first: 50) { nodes { id author { login } body createdAt } } } } } } } """ variables = {"owner": owner, "repo": repo, "pullNumber": pull_number} data = self.execute_graphql(query, variables) threads = ( data.get("repository", {}) .get("pullRequest", {}) .get("reviewThreads", {}) .get("nodes", []) ) # Filter by resolution status if requested if unresolved_only: threads = [t for t in threads if not t.get("isResolved", False)] return threads def add_thread_reply(self, pull_request_id: str, thread_id: str, body: str) -> dict[str, Any]: """ Add a reply to a review thread. Args: pull_request_id: Pull request node ID thread_id: Review thread node ID body: Reply content Returns: Created comment object """ query = """ mutation AddReply($pullRequestId: ID!, $threadId: ID!, $body: String!) { addPullRequestReviewThreadReply(input: { pullRequestId: $pullRequestId pullRequestReviewThreadId: $threadId body: $body }) { comment { id body createdAt author { login } } } } """ variables = {"pullRequestId": pull_request_id, "threadId": thread_id, "body": body} data = self.execute_graphql(query, variables) comment = data.get("addPullRequestReviewThreadReply", {}).get("comment", {}) if not comment: raise GitHubAPIError("Failed to create comment") return comment def resolve_thread(self, thread_id: str) -> dict[str, Any]: """ Resolve a review thread. Args: thread_id: Review thread node ID Returns: Updated thread object """ query = """ mutation ResolveThread($threadId: ID!) { resolveReviewThread(input: { threadId: $threadId }) { thread { id isResolved } } } """ variables = {"threadId": thread_id} data = self.execute_graphql(query, variables) thread = data.get("resolveReviewThread", {}).get("thread", {}) if not thread: raise GitHubAPIError("Failed to resolve thread") return thread