PR Review MCP Server

Instructions

Implementation Reference

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

  The handler function for the 'list_review_threads' MCP tool. It parses input arguments, fetches review threads using GitHubAPI, formats them into a structured JSON response including thread details and first comment info, 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:22-41 (schema)

  The input schema and metadata for the 'list_review_threads' tool, defining parameters: owner, repo, pull_number (required), and unresolved_only (optional boolean with default True).

      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)

  Tool dispatch registration in the call_tool handler, routing 'list_review_threads' calls to the specific handle_list_review_threads function.

  if name == "list_review_threads":
      return await handle_list_review_threads(api, arguments)

• src/pr_review_mcp/gh_api.py:99-159 (helper)

  Supporting GitHubAPI helper method that executes a GraphQL query to fetch review threads for a PR, including details like comments, and filters unresolved threads if specified.

  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