PR Review MCP Server

Instructions

Implementation Reference

• src/pr_review_mcp/tools.py:128-167 (handler)

  The handle_list_review_threads function that implements the core logic for the 'list_review_threads' tool: parses input arguments, calls GitHubAPI.list_review_threads, formats results into structured JSON, and returns as TextContent.
  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))]
• src/pr_review_mcp/tools.py:21-41 (schema)

  Defines the input schema, name, and description for the 'list_review_threads' tool.
  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"], }, ),
• src/pr_review_mcp/tools.py:111-112 (registration)

  Registration dispatch in the @server.call_tool() handler: routes calls to 'list_review_threads' to the specific handler function.
  if name == "list_review_threads": return await handle_list_review_threads(api, arguments)
• src/pr_review_mcp/gh_api.py:99-158 (helper)

  GitHubAPI.list_review_threads helper method: executes GraphQL query to fetch review threads for a PR, including comments, and optionally filters to unresolved threads.
  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