MCP Atlassian

"""Module for Jira search operations.""" import logging import requests from requests.exceptions import HTTPError from ..exceptions import MCPAtlassianAuthenticationError from ..models.jira import JiraSearchResult from .client import JiraClient from .constants import DEFAULT_READ_JIRA_FIELDS from .protocols import IssueOperationsProto logger = logging.getLogger("mcp-jira") class SearchMixin(JiraClient, IssueOperationsProto): """Mixin for Jira search operations.""" def search_issues( self, jql: str, fields: list[str] | tuple[str, ...] | set[str] | str | None = None, start: int = 0, limit: int = 50, expand: str | None = None, projects_filter: str | None = None, ) -> JiraSearchResult: """ Search for issues using JQL (Jira Query Language). Args: jql: JQL query string fields: Fields to return (comma-separated string, list, tuple, set, or "*all") start: Starting index if number of issues is greater than the limit Note: This parameter is ignored in Cloud environments and results will always start from the first page. limit: Maximum issues to return expand: Optional items to expand (comma-separated) projects_filter: Optional comma-separated list of project keys to filter by, overrides config Returns: JiraSearchResult object containing issues and metadata (total, start_at, max_results) Raises: MCPAtlassianAuthenticationError: If authentication fails with the Jira API (401/403) Exception: If there is an error searching for issues """ try: # Use projects_filter parameter if provided, otherwise fall back to config filter_to_use = projects_filter or self.config.projects_filter # Apply projects filter if present if filter_to_use: # Split projects filter by commas and handle possible whitespace projects = [p.strip() for p in filter_to_use.split(",")] # Build the project filter query part if len(projects) == 1: project_query = f'project = "{projects[0]}"' else: quoted_projects = [f'"{p}"' for p in projects] projects_list = ", ".join(quoted_projects) project_query = f"project IN ({projects_list})" # Add the project filter to existing query if not jql: # Empty JQL - just use project filter jql = project_query elif jql.strip().upper().startswith("ORDER BY"): # JQL starts with ORDER BY - prepend project filter jql = f"{project_query} {jql}" elif "project = " not in jql and "project IN" not in jql: # Only add if not already filtering by project jql = f"({jql}) AND {project_query}" logger.info(f"Applied projects filter to query: {jql}") # Convert fields to proper format if it's a list/tuple/set fields_param: str | None if fields is None: # Use default if None fields_param = ",".join(DEFAULT_READ_JIRA_FIELDS) elif isinstance(fields, list | tuple | set): fields_param = ",".join(fields) else: fields_param = fields if self.config.is_cloud: actual_total = -1 try: # Call 1: Get metadata (including total) using standard search API metadata_params = {"jql": jql, "maxResults": 0} metadata_response = self.jira.get( self.jira.resource_url("search"), params=metadata_params ) if ( isinstance(metadata_response, dict) and "total" in metadata_response ): try: actual_total = int(metadata_response["total"]) except (ValueError, TypeError): logger.warning( f"Could not parse 'total' from metadata response for JQL: {jql}. Received: {metadata_response.get('total')}" ) else: logger.warning( f"Could not retrieve total count from metadata response for JQL: {jql}. Response type: {type(metadata_response)}" ) except Exception as meta_err: logger.error( f"Error fetching metadata for JQL '{jql}': {str(meta_err)}" ) # Call 2: Get the actual issues using the enhanced method issues_response_list = self.jira.enhanced_jql_get_list_of_tickets( jql, fields=fields_param, limit=limit, expand=expand ) if not isinstance(issues_response_list, list): msg = f"Unexpected return value type from `jira.enhanced_jql_get_list_of_tickets`: {type(issues_response_list)}" logger.error(msg) raise TypeError(msg) response_dict_for_model = { "issues": issues_response_list, "total": actual_total, } search_result = JiraSearchResult.from_api_response( response_dict_for_model, base_url=self.config.url, requested_fields=fields_param, ) # Return the full search result object return search_result else: limit = min(limit, 50) response = self.jira.jql( jql, fields=fields_param, start=start, limit=limit, expand=expand ) if not isinstance(response, dict): msg = f"Unexpected return value type from `jira.jql`: {type(response)}" logger.error(msg) raise TypeError(msg) # Convert the response to a search result model search_result = JiraSearchResult.from_api_response( response, base_url=self.config.url, requested_fields=fields_param ) # Return the full search result object return search_result except HTTPError as http_err: if http_err.response is not None and http_err.response.status_code in [ 401, 403, ]: error_msg = ( f"Authentication failed for Jira API ({http_err.response.status_code}). " "Token may be expired or invalid. Please verify credentials." ) logger.error(error_msg) raise MCPAtlassianAuthenticationError(error_msg) from http_err else: logger.error(f"HTTP error during API call: {http_err}", exc_info=False) raise http_err except Exception as e: logger.error(f"Error searching issues with JQL '{jql}': {str(e)}") raise Exception(f"Error searching issues: {str(e)}") from e def get_board_issues( self, board_id: str, jql: str, fields: str | None = None, start: int = 0, limit: int = 50, expand: str | None = None, ) -> JiraSearchResult: """ Get all issues linked to a specific board. Args: board_id: The ID of the board jql: JQL query string fields: Fields to return (comma-separated string or "*all") start: Starting index limit: Maximum issues to return expand: Optional items to expand (comma-separated) Returns: JiraSearchResult object containing board issues and metadata Raises: Exception: If there is an error getting board issues """ try: # Determine fields_param fields_param = fields if fields_param is None: fields_param = ",".join(DEFAULT_READ_JIRA_FIELDS) response = self.jira.get_issues_for_board( board_id=board_id, jql=jql, fields=fields_param, start=start, limit=limit, expand=expand, ) if not isinstance(response, dict): msg = f"Unexpected return value type from `jira.get_issues_for_board`: {type(response)}" logger.error(msg) raise TypeError(msg) # Convert the response to a search result model search_result = JiraSearchResult.from_api_response( response, base_url=self.config.url, requested_fields=fields_param ) return search_result except requests.HTTPError as e: logger.error( f"Error searching issues for board with JQL '{board_id}': {str(e.response.content)}" ) raise Exception( f"Error searching issues for board with JQL: {str(e.response.content)}" ) from e except Exception as e: logger.error(f"Error searching issues for board with JQL '{jql}': {str(e)}") raise Exception( f"Error searching issues for board with JQL {str(e)}" ) from e def get_sprint_issues( self, sprint_id: str, fields: str | None = None, start: int = 0, limit: int = 50, ) -> JiraSearchResult: """ Get all issues linked to a specific sprint. Args: sprint_id: The ID of the sprint fields: Fields to return (comma-separated string or "*all") start: Starting index limit: Maximum issues to return Returns: JiraSearchResult object containing sprint issues and metadata Raises: Exception: If there is an error getting board issues """ try: # Determine fields_param fields_param = fields if fields_param is None: fields_param = ",".join(DEFAULT_READ_JIRA_FIELDS) response = self.jira.get_sprint_issues( sprint_id=sprint_id, start=start, limit=limit, ) if not isinstance(response, dict): msg = f"Unexpected return value type from `jira.get_sprint_issues`: {type(response)}" logger.error(msg) raise TypeError(msg) # Convert the response to a search result model search_result = JiraSearchResult.from_api_response( response, base_url=self.config.url, requested_fields=fields_param ) return search_result except requests.HTTPError as e: logger.error( f"Error searching issues for sprint '{sprint_id}': {str(e.response.content)}" ) raise Exception( f"Error searching issues for sprint: {str(e.response.content)}" ) from e except Exception as e: logger.error(f"Error searching issues for sprint: {sprint_id}': {str(e)}") raise Exception(f"Error searching issues for sprint: {str(e)}") from e