github-projects-mcp

Instructions

Args:
    owner: The GitHub organization or user name
    project_number: The project number
    limit: Maximum number of items to return (default: 50). When filtering, the system automatically fetches more items to improve efficiency.
    state: Optional state filter (e.g., "OPEN", "CLOSED"). Applies to Issues/PRs.
    filter_field_name: Optional custom field name to filter by (e.g., "Status"). Currently supports SingleSelect and Iteration fields.
    filter_field_value: Optional custom field value to filter by (e.g., "In Development"). Uses case-insensitive matching.
    cursor: Optional cursor for pagination. Use value from previous results to get next page.

Returns:
    A formatted string with item details.

Implementation Reference

• src/github_projects_mcp/server.py:115-278 (handler)

  The primary MCP tool handler and registration for 'get_project_items'. It is decorated with @mcp.tool(), validates inputs, delegates to GitHubClient.get_project_items, formats the response with item details, field values, pagination, and handles no-results debug info.

  @mcp.tool()
  async def get_project_items(
      owner: str,
      project_number: int,
      limit: int = 50,
      state: Optional[str] = None,
      filter_field_name: Optional[str] = None,
      filter_field_value: Optional[str] = None,
      cursor: Optional[str] = None,
  ) -> str:
      """Get items in a GitHub Project V2. Can filter by state OR a single custom field=value.
  
      Args:
          owner: The GitHub organization or user name
          project_number: The project number
          limit: Maximum number of items to return (default: 50). When filtering, the system automatically fetches more items to improve efficiency.
          state: Optional state filter (e.g., "OPEN", "CLOSED"). Applies to Issues/PRs.
          filter_field_name: Optional custom field name to filter by (e.g., "Status"). Currently supports SingleSelect and Iteration fields.
          filter_field_value: Optional custom field value to filter by (e.g., "In Development"). Uses case-insensitive matching.
          cursor: Optional cursor for pagination. Use value from previous results to get next page.
  
      Returns:
          A formatted string with item details.
      """
      if state and filter_field_name:
          return "Error: Cannot filter by both 'state' and a custom field ('filter_field_name') simultaneously."
  
      try:
          result = await github_client.get_project_items(
              owner,
              project_number,
              limit,
              state,
              filter_field_name,
              filter_field_value,
              cursor,
          )
  
          items = result["items"]
          page_info = result["pageInfo"]
          has_next_page = page_info.get("hasNextPage", False)
          end_cursor = page_info.get("endCursor")
  
          filter_desc = ""
          if state:
              filter_desc = f" (State: {state.upper()})"
          elif filter_field_name and filter_field_value:
              filter_desc = f" (Filter: {filter_field_name} = '{filter_field_value}')"
  
          pagination_desc = ""
          if cursor:
              pagination_desc = " (continued)"
  
          if not items:
              if cursor:
                  return f"No more items found in project #{project_number} for {owner}{filter_desc}"
              else:
                  # Add more context for debugging purposes when filtering returns no items
                  if filter_field_name and filter_field_value:
                      # Get available fields and options to help debug
                      try:
                          fields_details = await github_client.get_project_fields_details(
                              owner, project_number
                          )
                          field_info = None
  
                          # Try to find the field case-insensitively
                          for fname, finfo in fields_details.items():
                              if fname.lower() == filter_field_name.lower():
                                  field_info = finfo
                                  break
  
                          if field_info:
                              field_type = field_info.get("type", "Unknown")
                              if field_type == "ProjectV2SingleSelectField":
                                  available_options = list(
                                      field_info.get("options", {}).keys()
                                  )
                                  return (
                                      f"No items found in project #{project_number} for {owner}{filter_desc}\n\n"
                                      f"Debug info:\n"
                                      f"- Found field '{filter_field_name}' with type '{field_type}'\n"
                                      f"- Available options: {available_options}\n"
                                      f"- Note: Searched up to {limit} items. If the project has many items, some '{filter_field_value}' items might be beyond this scope.\n"
                                      f"- Tip: Try increasing the limit parameter or check if the field value spelling is correct."
                                  )
                              elif field_type == "ProjectV2IterationField":
                                  available_iterations = list(
                                      field_info.get("iterations", {}).keys()
                                  )
                                  return (
                                      f"No items found in project #{project_number} for {owner}{filter_desc}\n\n"
                                      f"Debug info:\n"
                                      f"- Found field '{filter_field_name}' with type '{field_type}'\n"
                                      f"- Available iterations: {available_iterations}\n"
                                      f"- Note: Searched up to {limit} items. If the project has many items, some '{filter_field_value}' items might be beyond this scope.\n"
                                      f"- Tip: Try increasing the limit parameter or check if the field value spelling is correct."
                                  )
                      except Exception as e:
                          logger.warning(
                              f"Could not get additional field details for debug info: {e}"
                          )
  
                  return f"No items found in project #{project_number} for {owner}{filter_desc}\n\nNote: If the project has many items, try increasing the limit parameter to search more thoroughly."
  
          # Format results
          result = f"Items in project #{project_number} for {owner}{filter_desc}{pagination_desc}:\n\n"
          for item in items:
              content = item.get("content", {})
              result += f"- Item ID: {item['id']}\n"
              item_type = content.get("__typename")
              repo_info = content.get("repository", {})
              repo_str = (
                  f"{repo_info.get('owner', {}).get('login')}/{repo_info.get('name')}"
                  if repo_info
                  else "N/A"
              )
  
              if item_type == "Issue":
                  result += f"  Type: Issue #{content.get('number')} ({repo_str})\n"
                  result += f"  Title: {content.get('title')}\n"
                  result += f"  State: {content.get('state')}\n"
                  result += f"  URL: {content.get('url')}\n"
              elif item_type == "PullRequest":
                  result += f"  Type: PR #{content.get('number')} ({repo_str})\n"
                  result += f"  Title: {content.get('title')}\n"
                  result += f"  State: {content.get('state')}\n"
                  result += f"  URL: {content.get('url')}\n"
              elif item_type == "DraftIssue":
                  # Include body for draft issues if available (check if client fetches it)
                  body = content.get("body", "")  # Assuming client might fetch body
                  result += f"  Type: Draft Issue ID: {content.get('id')}\n"
                  result += f"  Title: {content.get('title')}\n"
                  if body:
                      result += f"  Body: {body[:100]}...\n"  # Show preview
              else:
                  result += f"  Type: {item_type or 'Unknown'}\n"
                  result += f"  Content: {json.dumps(content)}\n"
  
              # Show processed field values
              if item.get("fieldValues"):
                  result += "  Field Values:\n"
                  for field_name, value in item["fieldValues"].items():
                      result += f"    - {field_name}: {value}\n"
              result += "\n"
  
          # Add pagination info
          if has_next_page and end_cursor:
              result += "\n--- Pagination ---\n"
              result += "More items available: Yes\n"
              result += f"Next page cursor: {end_cursor}\n"
              result += "To get the next page, use the cursor parameter:\n"
              result += f"cursor: {end_cursor}\n"
  
          return result
      except (
          GitHubClientError,
          ValueError,
      ) as e:  # Catch client errors and validation errors
          logger.error(
              f"Error getting items for project {owner}/{project_number} with filter: {e}"
          )
          return f"Error: Could not get items for project {owner}/{project_number}. Details: {e}"

• src/github_projects_mcp/github_client.py:384-757 (helper)

  The core helper function in GitHubClient that implements the logic to fetch project items via GraphQL, including dynamic query construction, field filtering (state or custom SingleSelect/Iteration fields with case-insensitive option matching), field value processing for multiple types, pagination support, and increased fetch limits for efficiency when filtering.

  async def get_project_items(
      self,
      owner: str,
      project_number: int,
      limit: int = 10,
      state: Optional[str] = None,
      filter_field_name: Optional[str] = None,
      filter_field_value: Optional[str] = None,
      cursor: Optional[str] = None,
  ) -> Dict[str, Any]:
      """
      Get items in a GitHub Project V2, optionally filtering by state or a custom field value.
      Args:
          owner: The GitHub organization or user name
          project_number: The project number
          limit: Maximum number of items to return per page (default: 10)
          state: Optional state to filter items by (e.g., "OPEN", "CLOSED").
          filter_field_name: Optional name of a custom field to filter by (e.g., "Status").
          filter_field_value: Optional value of the custom field to filter by (e.g., "Backlog").
          cursor: Optional cursor for pagination (default: None for first page)
      Returns:
          Dictionary containing:
              - items: List of project items
              - pageInfo: Information about pagination (hasNextPage, endCursor)
      Raises:
          GitHubClientError: If project or items cannot be retrieved, or filter is invalid.
          ValueError: If filter parameters are invalid.
      """
      try:
          project_id = await self.get_project_node_id(owner, project_number)
      except GitHubClientError as e:
          logger.error(f"Cannot get items: {e}")
          raise
  
      # When filtering, we need to fetch more items since many will be excluded
      # Increase the fetch size to be more efficient
      fetch_limit = limit
      if filter_field_name and filter_field_value:
          # Be more aggressive but respect GitHub's 100 record limit
          fetch_limit = min(max(limit * 5, 50), 100)
          logger.debug(
              f"Filtering enabled: increasing fetch limit from {limit} to {fetch_limit}"
          )
  
      # Prepare variables dict before use
      variables: Dict[str, Any] = {"projectId": project_id, "first": fetch_limit}
  
      # Add cursor if provided (for pagination)
      after_clause = ""
      if cursor:
          variables["cursor"] = cursor
          after_clause = ", after: $cursor"
  
      # Base query definition including fragments
      field_values_fragment = """
      fragment FieldValuesFragment on ProjectV2ItemFieldValueConnection {
          nodes {
             ... on ProjectV2ItemFieldTextValue { __typename text field { ... on ProjectV2FieldCommon { name } } }
             ... on ProjectV2ItemFieldDateValue { __typename date field { ... on ProjectV2FieldCommon { name } } }
             ... on ProjectV2ItemFieldSingleSelectValue { __typename name field { ... on ProjectV2FieldCommon { name } } }
             ... on ProjectV2ItemFieldNumberValue { __typename number field { ... on ProjectV2FieldCommon { name } } }
             ... on ProjectV2ItemFieldIterationValue { __typename title startDate duration field { ... on ProjectV2FieldCommon { name } } }
          }
      }
      """
      content_fragment = """
      fragment ContentFragment on ProjectV2ItemContent {
         ... on Issue { __typename id number title state url repository { name owner { login } } }
         ... on PullRequest { __typename id number title state url repository { name owner { login } } }
         ... on DraftIssue { __typename id title body }
      }
      """
  
      # Build filter parameters if needed
      filter_conditions = []
  
      if state:
          if state.upper() not in ["OPEN", "CLOSED"]:
              raise ValueError("Invalid state filter. Must be 'OPEN' or 'CLOSED'.")
          # For state filtering, let's collect all items and filter afterwards
          # as the API has changed how filtering works
  
      # Variables for field-based filtering
      field_id_var = None
      option_id_var = None
  
      if filter_field_name and filter_field_value:
          try:
              all_fields = await self.get_project_fields_details(
                  owner, project_number
              )
              logger.debug(f"All fields available: {list(all_fields.keys())}")
  
              # First try exact match
              field_info = all_fields.get(filter_field_name)
              # If not found, try case-insensitive match
              if not field_info:
                  actual_field_name = self._find_case_insensitive_key(
                      all_fields, filter_field_name
                  )
                  if actual_field_name:
                      field_info = all_fields.get(actual_field_name)
                      logger.info(
                          f"Found field '{actual_field_name}' using case-insensitive match for '{filter_field_name}'"
                      )
  
              if not field_info:
                  raise ValueError(f"Field '{filter_field_name}' not found.")
              field_id = field_info["id"]
              field_type = field_info["type"]
  
              logger.debug(
                  f"Found field '{filter_field_name}' with ID {field_id} and type {field_type}"
              )
  
              if field_type == "ProjectV2SingleSelectField":
                  available_options = list(field_info.get("options", {}).keys())
                  logger.debug(
                      f"Available options for '{filter_field_name}': {available_options}"
                  )
  
                  # First try exact match
                  option_id = field_info.get("options", {}).get(filter_field_value)
                  # If not found, try case-insensitive match
                  if not option_id:
                      options = field_info.get("options", {})
                      actual_option_name = self._find_case_insensitive_key(
                          options, filter_field_value
                      )
                      if actual_option_name:
                          option_id = options.get(actual_option_name)
                          logger.info(
                              f"Found option '{actual_option_name}' using case-insensitive match for '{filter_field_value}'"
                          )
  
                  if not option_id:
                      raise ValueError(
                          f"Option '{filter_field_value}' not found for field '{filter_field_name}'. Available: {available_options}"
                      )
                  field_id_var = field_id
                  option_id_var = option_id
                  logger.debug(
                      f"Using field ID {field_id_var} with option ID {option_id_var} for filtering"
                  )
              elif field_type == "ProjectV2IterationField":
                  available_iterations = list(field_info.get("iterations", {}).keys())
                  logger.debug(
                      f"Available iterations for '{filter_field_name}': {available_iterations}"
                  )
  
                  # First try exact match
                  iteration_id = field_info.get("iterations", {}).get(
                      filter_field_value
                  )
                  # If not found, try case-insensitive match
                  if not iteration_id:
                      iterations = field_info.get("iterations", {})
                      actual_iteration_name = self._find_case_insensitive_key(
                          iterations, filter_field_value
                      )
                      if actual_iteration_name:
                          iteration_id = iterations.get(actual_iteration_name)
                          logger.info(
                              f"Found iteration '{actual_iteration_name}' using case-insensitive match for '{filter_field_value}'"
                          )
  
                  if not iteration_id:
                      raise ValueError(
                          f"Iteration '{filter_field_value}' not found for field '{filter_field_name}'. Available: {available_iterations}"
                      )
                  field_id_var = field_id
                  option_id_var = iteration_id
                  logger.debug(
                      f"Using field ID {field_id_var} with iteration ID {option_id_var} for filtering"
                  )
              else:
                  logger.warning(
                      f"Filtering by field type '{field_type}' is not yet implemented."
                  )
          except GitHubClientError as e:
              logger.error(f"Error during field lookup for filtering: {e}")
              raise
          except ValueError as e:
              logger.error(f"Invalid filter input: {e}")
              raise
  
      # Use single curly braces for GraphQL, not double curly braces for f-string
      query = f"""
      {field_values_fragment}
      {content_fragment}
      query GetProjectItems($projectId: ID!, $first: Int!{', $cursor: String' if cursor else ''}) {{
        node(id: $projectId) {{
          ... on ProjectV2 {{
            items(first: $first{after_clause}) {{
              pageInfo {{
                hasNextPage
                endCursor
              }}
              nodes {{
                id
                type
                fieldValues(first: 20) {{ ...FieldValuesFragment }}
                content {{ ...ContentFragment }}
              }}
            }}
          }}
        }}
      }}
      """
  
      logger.debug(f"Executing items query: {query} with vars: {variables}")
  
      try:
          result = await self.execute_query(query, variables)
          if result is None:
              logger.warning(
                  f"Query returned None result for project {owner}/{project_number}"
              )
              return {
                  "items": [],
                  "pageInfo": {"hasNextPage": False, "endCursor": None},
              }
  
          items_data = result.get("node", {}).get("items")
          if items_data is None:  # Check if items key exists, even if null
              if result.get("node") is None:
                  raise GitHubClientError(
                      f"Project node not found for {owner}/{project_number}"
                  )
              else:
                  logger.info(
                      f"No items found matching filter criteria for project {owner}/{project_number}"
                  )
                  return {
                      "items": [],
                      "pageInfo": {"hasNextPage": False, "endCursor": None},
                  }
  
          # Get pagination info
          page_info = items_data.get("pageInfo", {})
          items = items_data.get("nodes", [])
  
          logger.debug(
              f"Retrieved {len(items)} items from project {owner}/{project_number}"
          )
  
          # Process field values
          filtered_items = []
          for item in items:
              if item.get("fieldValues") and item["fieldValues"].get("nodes"):
                  field_values = item["fieldValues"]["nodes"]
  
                  processed_values = {}
                  matches_field_filter = (
                      False if (field_id_var and option_id_var) else True
                  )
  
                  for fv in field_values:
                      raw_field_name = fv.get("field", {}).get("name")
                      # Sanitize the field name
                      if raw_field_name:
                          # Remove chars other than alphanumeric, space, underscore, hyphen
                          sanitized_field_name = re.sub(
                              r"[^\w\s-]", "", raw_field_name
                          ).strip()
                      else:
                          sanitized_field_name = "UnknownField"
  
                      field_name = (
                          sanitized_field_name or "UnnamedField"
                      )  # Ensure not empty
  
                      value = "N/A"
                      fv_type = fv.get("__typename")
                      if fv_type == "ProjectV2ItemFieldTextValue":
                          value = fv.get("text", "N/A")
                      elif fv_type == "ProjectV2ItemFieldDateValue":
                          value = fv.get("date", "N/A")
                      elif fv_type == "ProjectV2ItemFieldSingleSelectValue":
                          value = fv.get("name", "N/A")
                          # Check if this is the field we're filtering on
                          if (
                              field_id_var
                              and option_id_var
                              and field_name
                              and filter_field_name
                              and field_name.lower() == filter_field_name.lower()
                              and value
                              and filter_field_value
                              and value.lower() == filter_field_value.lower()
                          ):
                              matches_field_filter = True
                              logger.debug(
                                  f"Found matching item with field '{field_name}' = '{value}'"
                              )
                          elif (
                              field_id_var
                              and option_id_var
                              and field_name
                              and filter_field_name
                              and field_name.lower() == filter_field_name.lower()
                          ):
                              logger.debug(
                                  f"Field name matched but value did not: '{value}' != '{filter_field_value}'"
                              )
                      elif fv_type == "ProjectV2ItemFieldNumberValue":
                          value = fv.get("number", "N/A")
                      elif fv_type == "ProjectV2ItemFieldIterationValue":
                          title = fv.get("title", "N/A")
                          value = f"{title} (Start: {fv.get('startDate', 'N/A')})"
                          # Check if this is the field we're filtering on
                          if (
                              field_id_var
                              and option_id_var
                              and field_name
                              and filter_field_name
                              and field_name.lower() == filter_field_name.lower()
                              and title
                              and filter_field_value
                              and title.lower() == filter_field_value.lower()
                          ):
                              matches_field_filter = True
                              logger.debug(
                                  f"Found matching item with iteration field '{field_name}' = '{title}'"
                              )
                      processed_values[field_name] = value
  
                  item["fieldValues"] = processed_values
  
                  # Apply state filter if needed
                  matches_state_filter = True
                  if state and item.get("content"):
                      content_state = item["content"].get("state")
                      if content_state and content_state != state.upper():
                          matches_state_filter = False
  
                  if matches_field_filter and matches_state_filter:
                      filtered_items.append(item)
              else:
                  # Items without field values are included only if we're not doing field filtering
                  if not (field_id_var and option_id_var):
                      filtered_items.append(item)
  
          # When filtering, we may have fetched more than requested, so trim to the requested limit
          if filter_field_name and filter_field_value and len(filtered_items) > limit:
              filtered_items = filtered_items[:limit]
              # Update pagination info to indicate there may be more filtered results
              page_info = {
                  "hasNextPage": True,
                  "endCursor": page_info.get("endCursor"),
              }
  
          # Emergency check: if we're filtering and got very few results, warn that there might be more
          if (
              filter_field_name
              and filter_field_value
              and len(filtered_items) == 0
              and len(items) >= fetch_limit
              and fetch_limit < 100
          ):
              logger.warning(
                  f"Found 0 filtered items but fetched the maximum ({fetch_limit}). There might be more items beyond this limit. Consider increasing the search scope."
              )
  
          logger.debug(
              f"Filtered down to {len(filtered_items)} items for project {owner}/{project_number} using criteria field_name={filter_field_name}, field_value={filter_field_value}"
          )
          return {"items": filtered_items, "pageInfo": page_info}
      except GitHubClientError as e:
          logger.error(
              f"Failed to get items for project {owner}/{project_number}: {e}"
          )
          raise