MCP JIRA Server

"""Issue operations for JIRA.""" import logging from typing import Any, Dict, List, Optional from jira import JIRA, Issue as JiraIssue from jira.exceptions import JIRAError from mcp_jira.client import JiraClient logger = logging.getLogger(__name__) class IssueOperations: """Handles JIRA issue operations.""" def __init__(self, client: JiraClient): """Initialize issue operations. Args: client: JiraClient instance """ self.client = client def create_issue( self, project: str, summary: str, issue_type: str, description: Optional[str] = None, priority: Optional[str] = None, assignee: Optional[str] = None, labels: Optional[List[str]] = None, components: Optional[List[str]] = None, custom_fields: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> Dict[str, Any]: """Create a new JIRA issue. Args: project: Project key summary: Issue summary issue_type: Issue type name description: Issue description (optional) priority: Priority name (optional) assignee: Assignee account ID or username (optional) labels: List of labels (optional) components: List of component names (optional) custom_fields: Dictionary of custom field names/IDs to values (optional) **kwargs: Additional fields Returns: Dictionary with created issue information Raises: JIRAError: If issue creation fails """ try: jira = self.client.jira # Build fields dictionary fields: Dict[str, Any] = { "project": {"key": project}, "summary": summary, "issuetype": {"name": issue_type}, } if description: fields["description"] = description if priority: fields["priority"] = {"name": priority} if assignee: # Try account ID first, fallback to name if assignee.startswith("account"): fields["assignee"] = {"accountId": assignee} else: fields["assignee"] = {"name": assignee} if labels: fields["labels"] = labels if components: fields["components"] = [{"name": comp} for comp in components] # Add custom fields if custom_fields: prepared_custom = self.client.custom_fields.prepare_custom_fields(custom_fields) fields.update(prepared_custom) # Add any additional fields fields.update(kwargs) logger.info(f"Creating issue in project {project}") issue = jira.create_issue(fields=fields) return { "key": issue.key, "id": issue.id, "self": issue.self, "url": f"{self.client.config.jira_url}/browse/{issue.key}", } except JIRAError as e: logger.error(f"Failed to create issue: {e}") raise def get_issue( self, issue_key: str, fields: Optional[List[str]] = None, expand: Optional[str] = None, ) -> Dict[str, Any]: """Get issue details. Args: issue_key: Issue key (e.g., 'PROJ-123') fields: List of fields to retrieve (optional, defaults to all) expand: Comma-separated list of entities to expand (optional) Returns: Dictionary with issue details including custom fields Raises: JIRAError: If issue not found or retrieval fails """ try: jira = self.client.jira logger.info(f"Retrieving issue {issue_key}") issue = jira.issue(issue_key, fields=fields, expand=expand) return self._format_issue(issue) except JIRAError as e: logger.error(f"Failed to get issue {issue_key}: {e}") raise def update_issue( self, issue_key: str, fields: Optional[Dict[str, Any]] = None, custom_fields: Optional[Dict[str, Any]] = None, **kwargs: Any, ) -> Dict[str, Any]: """Update an existing issue. Args: issue_key: Issue key fields: Dictionary of standard fields to update custom_fields: Dictionary of custom fields to update **kwargs: Additional fields to update Returns: Dictionary with update result Raises: JIRAError: If update fails """ try: jira = self.client.jira update_fields = fields or {} # Prepare custom fields if custom_fields: prepared_custom = self.client.custom_fields.prepare_custom_fields(custom_fields) update_fields.update(prepared_custom) # Add additional fields update_fields.update(kwargs) logger.info(f"Updating issue {issue_key}") issue = jira.issue(issue_key) issue.update(fields=update_fields) return { "key": issue_key, "updated": True, "message": f"Successfully updated issue {issue_key}", } except JIRAError as e: logger.error(f"Failed to update issue {issue_key}: {e}") raise def delete_issue(self, issue_key: str) -> Dict[str, Any]: """Delete an issue. Args: issue_key: Issue key Returns: Dictionary with deletion result Raises: JIRAError: If deletion fails """ try: jira = self.client.jira logger.info(f"Deleting issue {issue_key}") issue = jira.issue(issue_key) issue.delete() return { "key": issue_key, "deleted": True, "message": f"Successfully deleted issue {issue_key}", } except JIRAError as e: logger.error(f"Failed to delete issue {issue_key}: {e}") raise def assign_issue( self, issue_key: str, assignee: Optional[str] = None ) -> Dict[str, Any]: """Assign issue to a user. Args: issue_key: Issue key assignee: Account ID, username, or None to unassign Returns: Dictionary with assignment result Raises: JIRAError: If assignment fails """ try: jira = self.client.jira logger.info(f"Assigning issue {issue_key} to {assignee or 'unassigned'}") jira.assign_issue(issue_key, assignee) return { "key": issue_key, "assigned": True, "assignee": assignee, "message": f"Successfully assigned issue {issue_key}", } except JIRAError as e: logger.error(f"Failed to assign issue {issue_key}: {e}") raise def add_watcher(self, issue_key: str, watcher: str) -> Dict[str, Any]: """Add a watcher to an issue. Args: issue_key: Issue key watcher: Account ID or username Returns: Dictionary with result Raises: JIRAError: If operation fails """ try: jira = self.client.jira logger.info(f"Adding watcher {watcher} to issue {issue_key}") jira.add_watcher(issue_key, watcher) return { "key": issue_key, "watcher_added": True, "watcher": watcher, } except JIRAError as e: logger.error(f"Failed to add watcher to issue {issue_key}: {e}") raise def _format_issue(self, issue: JiraIssue) -> Dict[str, Any]: """Format JIRA issue object to dictionary. Args: issue: JIRA issue object Returns: Formatted dictionary """ fields = issue.raw.get("fields", {}) # Extract custom fields custom = self.client.custom_fields.extract_custom_fields(fields) return { "key": issue.key, "id": issue.id, "self": issue.self, "url": f"{self.client.config.jira_url}/browse/{issue.key}", "summary": fields.get("summary"), "description": fields.get("description"), "status": fields.get("status", {}).get("name"), "priority": fields.get("priority", {}).get("name") if fields.get("priority") else None, "assignee": fields.get("assignee", {}).get("displayName") if fields.get("assignee") else None, "reporter": fields.get("reporter", {}).get("displayName") if fields.get("reporter") else None, "created": fields.get("created"), "updated": fields.get("updated"), "labels": fields.get("labels", []), "components": [c.get("name") for c in fields.get("components", [])], "custom_fields": custom, "raw_fields": fields, }