github-projects-mcp

Instructions

Implementation Reference

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

  MCP tool handler: processes args, invokes GitHub client, handles errors, formats items with repo info, field values, pagination 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)

  Core helper in GitHubClient: executes GraphQL query for project items, handles filtering by state or custom fields (SingleSelect, Iteration), processes field values into readable format, supports pagination.
  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
• src/github_projects_mcp/server.py:115-115 (registration)

  Tool registration via FastMCP @mcp.tool() decorator on the handler function.
  @mcp.tool()
• src/github_projects_mcp/server.py:116-124 (schema)

  Input schema defined by function parameters and docstring; output is formatted string.
  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: