Jira MCP Server

import asyncio import json import logging import os import sys from enum import Enum from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Union # --- Setup a dedicated file logger --- log_file_path = Path(__file__).parent / "jira_mcp_debug.log" logger = logging.getLogger("JiraMCPLogger") logger.setLevel(logging.DEBUG) # Capture all levels of logs # Create a file handler to write logs to a file # Use 'w' to overwrite the file on each run, ensuring a clean log handler = logging.FileHandler(log_file_path, mode="w") handler.setLevel(logging.DEBUG) # Create a formatter to make the logs readable formatter = logging.Formatter( "%(asctime)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s" ) handler.setFormatter(formatter) # Add the handler to the logger if not logger.handlers: logger.addHandler(handler) logger.info("Logger initialized. All subsequent logs will go to jira_mcp_debug.log") # --- End of logger setup --- try: from jira import JIRA except ImportError: from .jira import JIRA from pydantic import BaseModel from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import EmbeddedResource, ImageContent, TextContent, Tool from .jira_v3_api import JiraV3APIClient try: from dotenv import load_dotenv # Try to load from .env file if it exists env_path = Path(__file__).parent.parent.parent.parent / ".env" if env_path.exists(): load_dotenv(dotenv_path=env_path) except ImportError: # dotenv is optional pass class JiraTools(str, Enum): GET_PROJECTS = "get_jira_projects" GET_ISSUE = "get_jira_issue" SEARCH_ISSUES = "search_jira_issues" CREATE_ISSUE = "create_jira_issue" CREATE_ISSUES = "create_jira_issues" ADD_COMMENT = "add_jira_comment" GET_TRANSITIONS = "get_jira_transitions" TRANSITION_ISSUE = "transition_jira_issue" CREATE_PROJECT = "create_jira_project" GET_PROJECT_ISSUE_TYPES = "get_jira_project_issue_types" class JiraIssueField(BaseModel): name: str value: str class JiraIssueResult(BaseModel): key: str summary: str description: Optional[str] = None status: Optional[str] = None assignee: Optional[str] = None reporter: Optional[str] = None created: Optional[str] = None updated: Optional[str] = None fields: Optional[Dict[str, Any]] = None comments: Optional[List[Dict[str, Any]]] = None watchers: Optional[Dict[str, Any]] = None attachments: Optional[List[Dict[str, Any]]] = None subtasks: Optional[List[Dict[str, Any]]] = None project: Optional[Dict[str, Any]] = None issue_links: Optional[List[Dict[str, Any]]] = None worklog: Optional[List[Dict[str, Any]]] = None timetracking: Optional[Dict[str, Any]] = None class JiraProjectResult(BaseModel): key: str name: str id: str lead: Optional[str] = None class JiraTransitionResult(BaseModel): id: str name: str class JiraServer: def __init__( self, server_url: str = None, auth_method: str = None, username: str = None, password: str = None, token: str = None, ): self.server_url = server_url self.auth_method = auth_method self.username = username self.password = password self.token = token self._v3_api_client = JiraV3APIClient( server_url=self.server_url, username=self.username, token=self.token, password=password, ) self.client = None def connect(self): """Connect to Jira server using provided authentication details""" if not self.server_url: print("Error: Jira server URL not provided") return False error_messages = [] # Try multiple auth methods if possible try: # First, try the specified auth method if self.auth_method == "basic_auth": # Basic auth - either username/password or username/token if self.username and self.password: try: print(f"Trying basic_auth with username and password") self.client = JIRA( server=self.server_url, basic_auth=(self.username, self.password), ) print("Connection successful with username/password") return True except Exception as e: error_msg = f"Failed basic_auth with username/password: {type(e).__name__}: {str(e)}" print(error_msg) error_messages.append(error_msg) if self.username and self.token: try: print(f"Trying basic_auth with username and API token") self.client = JIRA( server=self.server_url, basic_auth=(self.username, self.token), ) print("Connection successful with username/token") return True except Exception as e: error_msg = f"Failed basic_auth with username/token: {type(e).__name__}: {str(e)}" print(error_msg) error_messages.append(error_msg) print("Error: Username and password/token required for basic auth") error_messages.append( "Username and password/token required for basic auth" ) elif self.auth_method == "token_auth": # Token auth - just need the token if self.token: try: print(f"Trying token_auth with token") self.client = JIRA( server=self.server_url, token_auth=self.token ) print("Connection successful with token_auth") return True except Exception as e: error_msg = f"Failed token_auth: {type(e).__name__}: {str(e)}" print(error_msg) error_messages.append(error_msg) else: print("Error: Token required for token auth") error_messages.append("Token required for token auth") # If we're here and have a token, try using it with basic_auth for Jira Cloud # (even if auth_method wasn't basic_auth) if self.token and self.username and not self.client: try: print(f"Trying fallback to basic_auth with username and token") self.client = JIRA( server=self.server_url, basic_auth=(self.username, self.token) ) print("Connection successful with fallback basic_auth") return True except Exception as e: error_msg = ( f"Failed fallback to basic_auth: {type(e).__name__}: {str(e)}" ) print(error_msg) error_messages.append(error_msg) # If we're here and have a token, try using token_auth as a fallback # (even if auth_method wasn't token_auth) if self.token and not self.client: try: print(f"Trying fallback to token_auth") self.client = JIRA(server=self.server_url, token_auth=self.token) print("Connection successful with fallback token_auth") return True except Exception as e: error_msg = ( f"Failed fallback to token_auth: {type(e).__name__}: {str(e)}" ) print(error_msg) error_messages.append(error_msg) # Last resort: try anonymous access try: print(f"Trying anonymous access as last resort") self.client = JIRA(server=self.server_url) print("Connection successful with anonymous access") return True except Exception as e: error_msg = f"Failed anonymous access: {type(e).__name__}: {str(e)}" print(error_msg) error_messages.append(error_msg) # If we got here, all connection attempts failed print(f"All connection attempts failed: {', '.join(error_messages)}") return False except Exception as e: error_msg = f"Unexpected error in connect(): {type(e).__name__}: {str(e)}" print(error_msg) error_messages.append(error_msg) return False def _get_v3_api_client(self) -> JiraV3APIClient: """Get or create a v3 API client instance""" if not self._v3_api_client: self._v3_api_client = JiraV3APIClient( server_url=self.server_url, username=self.username, password=self.password, token=self.token, ) return self._v3_api_client async def get_jira_projects(self) -> List[JiraProjectResult]: """Get all accessible Jira projects using v3 REST API""" logger.info("Starting get_jira_projects...") all_projects_data = [] start_at = 0 max_results = 50 page_count = 0 while True: page_count += 1 logger.info( f"Pagination loop, page {page_count}: startAt={start_at}, maxResults={max_results}" ) try: response = await self._v3_api_client.get_projects( start_at=start_at, max_results=max_results ) projects = response.get("values", []) if not projects: logger.info("No more projects returned. Breaking pagination loop.") break all_projects_data.extend(projects) if response.get("isLast", False): logger.info("'isLast' is True. Breaking pagination loop.") break start_at += len(projects) # Yield control to the event loop to prevent deadlocks in the MCP framework. await asyncio.sleep(0) except Exception as e: logger.error( "Error inside get_jira_projects pagination loop", exc_info=True ) raise logger.info( f"Finished get_jira_projects. Total projects found: {len(all_projects_data)}" ) results = [] for p in all_projects_data: results.append( JiraProjectResult( key=p.get("key"), name=p.get("name"), id=str(p.get("id")), lead=(p.get("lead") or {}).get("displayName"), ) ) logger.info(f"Added project {p.get('key')} to results") logger.info(f"Returning {len(results)} projects") sys.stdout.flush() # Flush stdout to ensure it's sent to MCP, otherwise hang occurs return results def get_jira_issue(self, issue_key: str) -> JiraIssueResult: """Get details for a specific issue by key""" if not self.client: if not self.connect(): # Connection failed - provide clear error message raise ValueError( f"Failed to connect to Jira server at {self.server_url}. Check your authentication credentials." ) try: issue = self.client.issue(issue_key) # Extract comments if available comments = [] if hasattr(issue.fields, "comment") and hasattr( issue.fields.comment, "comments" ): for comment in issue.fields.comment.comments: comments.append( { "author": ( getattr( comment.author, "displayName", str(comment.author) ) if hasattr(comment, "author") else "Unknown" ), "body": comment.body, "created": comment.created, } ) # Create a fields dictionary with custom fields fields = {} for field_name in dir(issue.fields): if not field_name.startswith("_") and field_name not in [ "comment", "attachment", "summary", "description", "status", "assignee", "reporter", "created", "updated", ]: value = getattr(issue.fields, field_name) if value is not None: # Handle special field types if hasattr(value, "name"): fields[field_name] = value.name elif hasattr(value, "value"): fields[field_name] = value.value elif isinstance(value, list): if len(value) > 0: if hasattr(value[0], "name"): fields[field_name] = [item.name for item in value] else: fields[field_name] = value else: fields[field_name] = str(value) return JiraIssueResult( key=issue.key, summary=issue.fields.summary, description=issue.fields.description, status=( issue.fields.status.name if hasattr(issue.fields, "status") else None ), assignee=( issue.fields.assignee.displayName if hasattr(issue.fields, "assignee") and issue.fields.assignee else None ), reporter=( issue.fields.reporter.displayName if hasattr(issue.fields, "reporter") and issue.fields.reporter else None ), created=( issue.fields.created if hasattr(issue.fields, "created") else None ), updated=( issue.fields.updated if hasattr(issue.fields, "updated") else None ), fields=fields, comments=comments, ) except Exception as e: print(f"Failed to get issue {issue_key}: {type(e).__name__}: {str(e)}") raise ValueError( f"Failed to get issue {issue_key}: {type(e).__name__}: {str(e)}" ) async def search_jira_issues( self, jql: str, max_results: int = 10 ) -> List[JiraIssueResult]: """Search for issues using JQL via v3 REST API with pagination support""" logger.info("Starting search_jira_issues...") try: # Use v3 API client v3_client = self._get_v3_api_client() # Collect all issues from all pages all_issues = [] start_at = 0 page_size = min(max_results, 100) # Jira typically limits to 100 per page while True: logger.debug(f"Fetching page starting at {start_at} with page size {page_size}") response_data = await v3_client.search_issues( jql=jql, start_at=start_at, max_results=page_size ) # Extract issues from current page page_issues = response_data.get("issues", []) all_issues.extend(page_issues) logger.debug(f"Retrieved {len(page_issues)} issues from current page. Total so far: {len(all_issues)}") # Check if we've reached the user's max_results limit if len(all_issues) >= max_results: # Trim to exact max_results if we exceeded it all_issues = all_issues[:max_results] logger.debug(f"Reached max_results limit of {max_results}, stopping pagination") break # Check if this is the last page according to API is_last = response_data.get("isLast", True) if is_last: logger.debug("API indicates this is the last page, stopping pagination") break # If we have more pages, prepare for next iteration start_at = len(all_issues) # Use actual number of issues retrieved so far # Adjust page size for next request to not exceed max_results remaining_needed = max_results - len(all_issues) page_size = min(remaining_needed, 100) # Return raw issues list for full JSON data logger.info(f"Returning raw issues ({len(all_issues)}) for JQL: {jql}") return all_issues except Exception as e: error_msg = f"Failed to search issues: {type(e).__name__}: {str(e)}" logger.error(error_msg, exc_info=True) print(error_msg) raise ValueError(error_msg) async def create_jira_issue( self, project: str, summary: str, description: str, issue_type: str, fields: Optional[Dict[str, Any]] = None, ) -> JiraIssueResult: """Create a new Jira issue using v3 REST API Args: project: Project key (e.g., 'PROJ') summary: Issue summary/title description: Issue description issue_type: Issue type - common values include 'Bug', 'Task', 'Story', 'Epic', 'New Feature', 'Improvement' Note: Available issue types vary by Jira instance and project fields: Optional additional fields dictionary Returns: JiraIssueResult object with the created issue details Example: # Create a bug await create_jira_issue( project='PROJ', summary='Login button not working', description='The login button on the homepage is not responding to clicks', issue_type='Bug' ) # Create a task with custom fields await create_jira_issue( project='PROJ', summary='Update documentation', description='Update API documentation with new endpoints', issue_type='Task', fields={ 'assignee': 'jsmith', 'labels': ['documentation', 'api'], 'priority': {'name': 'High'} } ) """ logger.info("Starting create_jira_issue...") try: # Create a properly formatted issue dictionary issue_dict = {} # Process required fields first # Project field - required if isinstance(project, str): issue_dict["project"] = {"key": project} else: issue_dict["project"] = project # Summary - required issue_dict["summary"] = summary # Description if description: issue_dict["description"] = description # Issue type - required, with validation for common issue types logger.info( f"Processing issue_type: '{issue_type}' (type: {type(issue_type)})" ) common_types = [ "bug", "task", "story", "epic", "improvement", "newfeature", "new feature", ] if isinstance(issue_type, str): # Check for common issue type variants and fix case-sensitivity issues issue_type_lower = issue_type.lower() if issue_type_lower in common_types: # Convert first letter to uppercase for standard Jira types issue_type_proper = issue_type_lower.capitalize() if ( issue_type_lower == "new feature" or issue_type_lower == "newfeature" ): issue_type_proper = "New Feature" logger.info( f"Note: Converting issue type from '{issue_type}' to '{issue_type_proper}'" ) issue_dict["issuetype"] = {"name": issue_type_proper} else: # Use the type as provided - some Jira instances have custom types issue_dict["issuetype"] = {"name": issue_type} else: issue_dict["issuetype"] = issue_type # Add any additional fields with proper type handling if fields: for key, value in fields.items(): # Skip fields we've already processed if key in [ "project", "summary", "description", "issuetype", "issue_type", ]: continue # Handle special fields that require specific formats if key == "assignees" or key == "assignee": # Convert string to array for assignees or proper format for assignee if isinstance(value, str): if key == "assignees": issue_dict[key] = [value] if value else [] else: # assignee issue_dict[key] = {"name": value} if value else None elif isinstance(value, list) and key == "assignee" and value: # If assignee is a list but should be a dict with name issue_dict[key] = {"name": value[0]} else: issue_dict[key] = value elif key == "labels": # Convert string to array for labels if isinstance(value, str): issue_dict[key] = [value] if value else [] else: issue_dict[key] = value elif key == "milestone": # Convert string to number for milestone if isinstance(value, str) and value.isdigit(): issue_dict[key] = int(value) else: issue_dict[key] = value else: issue_dict[key] = value # Use v3 API client v3_client = self._get_v3_api_client() response_data = await v3_client.create_issue(fields=issue_dict) # Extract issue details from v3 API response issue_key = response_data.get("key") issue_id = response_data.get("id") logger.info(f"Successfully created issue {issue_key} (ID: {issue_id})") # Return JiraIssueResult with the created issue details # For v3 API, we return what we have from the create response return JiraIssueResult( key=issue_key, summary=summary, # Use the summary we provided description=description, # Use the description we provided status="Open", # Default status for new issues ) except Exception as e: error_msg = f"Failed to create issue: {type(e).__name__}: {str(e)}" logger.error(error_msg, exc_info=True) # Enhanced error handling for issue type errors if "issuetype" in str(e).lower() or "issue type" in str(e).lower(): logger.info( "Issue type error detected, trying to provide helpful suggestions..." ) try: project_key = ( project if isinstance(project, str) else project.get("key") ) if project_key: issue_types = await self.get_jira_project_issue_types( project_key ) type_names = [t.get("name") for t in issue_types] logger.info( f"Available issue types for project {project_key}: {', '.join(type_names)}" ) # Try to find the closest match attempted_type = issue_type closest = None attempted_lower = attempted_type.lower() for t in type_names: if ( attempted_lower in t.lower() or t.lower() in attempted_lower ): closest = t break if closest: logger.info( f"The closest match to '{attempted_type}' is '{closest}'" ) error_msg += f" Available types: {', '.join(type_names)}. Closest match: '{closest}'" else: error_msg += f" Available types: {', '.join(type_names)}" except Exception as fetch_error: logger.error(f"Could not fetch issue types: {str(fetch_error)}") raise ValueError(error_msg) # Re-raise the exception with more details if "issuetype" in error_message.lower(): raise ValueError( f"Invalid issue type '{issue_dict.get('issuetype', {}).get('name', 'Unknown')}'. " + "Use get_jira_project_issue_types(project_key) to get valid types." ) raise return JiraIssueResult( key=new_issue.key, summary=new_issue.fields.summary, description=new_issue.fields.description, status=( new_issue.fields.status.name if hasattr(new_issue.fields, "status") else None ), ) except Exception as e: print(f"Failed to create issue: {type(e).__name__}: {str(e)}") raise ValueError(f"Failed to create issue: {type(e).__name__}: {str(e)}") async def create_jira_issues( self, field_list: List[Dict[str, Any]], prefetch: bool = True ) -> List[Dict[str, Any]]: """Bulk create new Jira issues using v3 REST API. Parameters: field_list (List[Dict[str, Any]]): a list of dicts each containing field names and the values to use. Each dict is an individual issue to create. prefetch (bool): True reloads the created issue Resource so all of its data is present in the value returned (Default: True) Returns: List[Dict[str, Any]]: List of created issues with their details Issue Types: Common issue types include: 'Bug', 'Task', 'Story', 'Epic', 'New Feature', 'Improvement' Note: Available issue types vary by Jira instance and project Example: # Create multiple issues in bulk await create_jira_issues([ { 'project': 'PROJ', 'summary': 'Implement user authentication', 'description': 'Add login and registration functionality', 'issue_type': 'Story' # Note: case-sensitive, match to your Jira instance types }, { 'project': 'PROJ', 'summary': 'Fix navigation bar display on mobile', 'description': 'Navigation bar is not displaying correctly on mobile devices', 'issue_type': 'Bug', 'priority': {'name': 'High'}, 'labels': ['mobile', 'ui'] } ]) """ logger.info("Starting create_jira_issues...") try: # Process each field dict to ensure proper formatting for v3 API processed_field_list = [] for fields in field_list: # Create a properly formatted issue dictionary issue_dict = {} # Process required fields first to ensure they exist # Project field - required if "project" not in fields: raise ValueError("Each issue must have a 'project' field") project_value = fields["project"] if isinstance(project_value, str): issue_dict["project"] = {"key": project_value} else: issue_dict["project"] = project_value # Summary field - required if "summary" not in fields: raise ValueError("Each issue must have a 'summary' field") issue_dict["summary"] = fields["summary"] # Description field - convert to ADF format for v3 API if it's a simple string if "description" in fields: description = fields["description"] if isinstance(description, str): # Convert simple string to Atlassian Document Format issue_dict["description"] = { "type": "doc", "version": 1, "content": [ { "type": "paragraph", "content": [ { "type": "text", "text": description } ] } ] } else: # Assume it's already in ADF format issue_dict["description"] = description # Issue type field - required, handle both 'issuetype' and 'issue_type' issue_type = None if "issuetype" in fields: issue_type = fields["issuetype"] elif "issue_type" in fields: issue_type = fields["issue_type"] else: raise ValueError( "Each issue must have an 'issuetype' or 'issue_type' field" ) # Check for common issue type variants and fix case-sensitivity issues logger.debug( f"Processing bulk issue_type: '{issue_type}' (type: {type(issue_type)})" ) common_types = [ "bug", "task", "story", "epic", "improvement", "newfeature", "new feature", ] if isinstance(issue_type, str): issue_type_lower = issue_type.lower() if issue_type_lower in common_types: # Convert first letter to uppercase for standard Jira types issue_type_proper = issue_type_lower.capitalize() if ( issue_type_lower == "new feature" or issue_type_lower == "newfeature" ): issue_type_proper = "New Feature" logger.debug( f"Converting issue type from '{issue_type}' to '{issue_type_proper}'" ) issue_dict["issuetype"] = {"name": issue_type_proper} else: # Use the type as provided - some Jira instances have custom types issue_dict["issuetype"] = {"name": issue_type} else: issue_dict["issuetype"] = issue_type # Process other fields for key, value in fields.items(): if key in [ "project", "summary", "description", "issuetype", "issue_type", ]: # Skip fields we've already processed continue # Handle special fields that require specific formats if key == "assignees" or key == "assignee": # Convert string to array for assignees or proper format for assignee if isinstance(value, str): if key == "assignees": issue_dict[key] = [value] if value else [] else: # assignee issue_dict[key] = {"name": value} if value else None elif isinstance(value, list) and key == "assignee" and value: # If assignee is a list but should be a dict with name issue_dict[key] = {"name": value[0]} else: issue_dict[key] = value elif key == "labels": # Convert string to array for labels if isinstance(value, str): issue_dict[key] = [value] if value else [] else: issue_dict[key] = value elif key == "milestone": # Convert string to number for milestone if isinstance(value, str) and value.isdigit(): issue_dict[key] = int(value) else: issue_dict[key] = value else: issue_dict[key] = value # Add to the field list in v3 API format processed_field_list.append({"fields": issue_dict}) logger.debug(f"Processed field list: {json.dumps(processed_field_list, indent=2)}") # Use v3 API client v3_client = self._get_v3_api_client() # Call the bulk create API response_data = await v3_client.bulk_create_issues(processed_field_list) # Process the results to maintain compatibility with existing interface processed_results = [] # Handle successful issues if "issues" in response_data: for issue in response_data["issues"]: processed_results.append({ "key": issue.get("key"), "id": issue.get("id"), "self": issue.get("self"), "success": True, }) # Handle errors if "errors" in response_data: for error in response_data["errors"]: processed_results.append({ "error": error, "success": False, }) logger.info(f"Successfully processed {len(processed_results)} issue creations") return processed_results except Exception as e: error_msg = f"Failed to create issues in bulk: {type(e).__name__}: {str(e)}" logger.error(error_msg, exc_info=True) print(error_msg) raise ValueError(error_msg) async def add_jira_comment(self, issue_key: str, comment: str) -> Dict[str, Any]: """Add a comment to an issue using v3 REST API""" logger.info("Starting add_jira_comment...") try: # Use v3 API client v3_client = self._get_v3_api_client() comment_result = await v3_client.add_comment( issue_id_or_key=issue_key, comment=comment, ) # Extract useful information from the v3 API response response_data = { "id": comment_result.get("id"), "body": comment_result.get("body", {}), "created": comment_result.get("created"), "updated": comment_result.get("updated"), } # Extract author information if available if "author" in comment_result: author = comment_result["author"] response_data["author"] = author.get("displayName", "Unknown") else: response_data["author"] = "Unknown" logger.info(f"Successfully added comment to issue {issue_key}") return response_data except Exception as e: error_msg = ( f"Failed to add comment to {issue_key}: {type(e).__name__}: {str(e)}" ) logger.error(error_msg, exc_info=True) print(error_msg) raise ValueError(error_msg) async def get_jira_transitions(self, issue_key: str) -> List[JiraTransitionResult]: """Get available transitions for an issue using v3 REST API""" logger.info("Starting get_jira_transitions...") try: # Use v3 API client v3_client = self._get_v3_api_client() response_data = await v3_client.get_transitions(issue_id_or_key=issue_key) # Extract transitions from response transitions = response_data.get("transitions", []) # Convert to JiraTransitionResult objects maintaining compatibility results = [ JiraTransitionResult(id=transition["id"], name=transition["name"]) for transition in transitions ] logger.info(f"Found {len(results)} transitions for issue {issue_key}") return results except Exception as e: error_msg = f"Failed to get transitions for {issue_key}: {type(e).__name__}: {str(e)}" logger.error(error_msg, exc_info=True) print(error_msg) raise ValueError(error_msg) async def transition_jira_issue( self, issue_key: str, transition_id: str, comment: Optional[str] = None, fields: Optional[Dict[str, Any]] = None, ) -> bool: """Transition an issue to a new state using v3 REST API""" logger.info("Starting transition_jira_issue...") try: # Use v3 API client v3_client = self._get_v3_api_client() await v3_client.transition_issue( issue_id_or_key=issue_key, transition_id=transition_id, fields=fields, comment=comment, ) logger.info( f"Successfully transitioned issue {issue_key} to transition {transition_id}" ) return True except Exception as e: error_msg = ( f"Failed to transition {issue_key}: {type(e).__name__}: {str(e)}" ) logger.error(error_msg, exc_info=True) print(error_msg) raise ValueError(error_msg) async def get_jira_project_issue_types( self, project_key: str ) -> List[Dict[str, Any]]: """Get all available issue types for a specific project using v3 REST API Args: project_key: The project key (e.g., 'PROJ') - kept for backward compatibility, but the new API returns all issue types for the user Returns: List of issue type dictionaries with name, id, and description Example: get_jira_project_issue_types('PROJ') # Returns all issue types accessible to user """ logger.info("Starting get_jira_project_issue_types...") try: # Use v3 API client to get all issue types v3_client = self._get_v3_api_client() response_data = await v3_client.get_issue_types() # The new API returns the issue types directly as a list, not wrapped in an object issue_types_data = ( response_data if isinstance(response_data, list) else response_data.get("issueTypes", []) ) # Convert to the expected format maintaining compatibility issue_types = [] for issuetype in issue_types_data: issue_types.append( { "id": issuetype.get("id"), "name": issuetype.get("name"), "description": issuetype.get("description"), } ) logger.info( f"Found {len(issue_types)} issue types (project_key: {project_key})" ) return issue_types except Exception as e: error_msg = f"Failed to get issue types: {type(e).__name__}: {str(e)}" logger.error(error_msg, exc_info=True) print(error_msg) raise ValueError(error_msg) async def create_jira_project( self, key: str, name: Optional[str] = None, assignee: Optional[str] = None, ptype: str = "software", template_name: Optional[str] = None, avatarId: Optional[int] = None, issueSecurityScheme: Optional[int] = None, permissionScheme: Optional[int] = None, projectCategory: Optional[int] = None, notificationScheme: Optional[int] = None, categoryId: Optional[int] = None, url: str = "", ) -> JiraProjectResult: """Create a project using Jira's v3 REST API Args: key: Project key (required) - must match Jira project key requirements name: Project name (defaults to key if not provided) assignee: Lead account ID or username ptype: Project type key ('software', 'business', 'service_desk') template_name: Project template key for creating from templates avatarId: ID of the avatar to use for the project issueSecurityScheme: ID of the issue security scheme permissionScheme: ID of the permission scheme projectCategory: ID of the project category notificationScheme: ID of the notification scheme categoryId: Same as projectCategory (alternative parameter) url: URL for project information/documentation Returns: JiraProjectResult with the created project details Note: This method uses Jira's v3 REST API endpoint: POST /rest/api/3/project Example: # Create a basic software project create_jira_project( key='PROJ', name='My Project', ptype='software' ) # Create with template create_jira_project( key='BUSI', name='Business Project', ptype='business', template_name='com.atlassian.jira-core-project-templates:jira-core-simplified-task-tracking' ) """ if not key: raise ValueError("Project key is required") try: # Get the v3 API client v3_client = self._get_v3_api_client() # Create project using v3 API response_data = await v3_client.create_project( key=key, name=name, assignee=assignee, ptype=ptype, template_name=template_name, avatarId=avatarId, issueSecurityScheme=issueSecurityScheme, permissionScheme=permissionScheme, projectCategory=projectCategory, notificationScheme=notificationScheme, categoryId=categoryId, url=url, ) # Extract project details from response project_id = response_data.get("id", "0") project_key = response_data.get("key", key) # For lead information, we would need to make another API call # For now, return None for lead as it's optional in our result model lead = None return JiraProjectResult( key=project_key, name=name or key, id=str(project_id), lead=lead ) except Exception as e: error_msg = str(e) print(f"Error creating project with v3 API: {error_msg}") raise ValueError(f"Error creating project: {error_msg}") async def serve( server_url: Optional[str] = None, auth_method: Optional[str] = None, username: Optional[str] = None, password: Optional[str] = None, token: Optional[str] = None, ) -> None: server = Server("mcp-jira") jira_server = JiraServer( server_url=server_url, auth_method=auth_method, username=username, password=password, token=token, ) @server.list_tools() async def list_tools() -> list[Tool]: """List available Jira tools.""" return [ Tool( name=JiraTools.GET_PROJECTS.value, description="Get all accessible Jira projects", inputSchema={"type": "object", "properties": {}, "required": []}, ), Tool( name=JiraTools.GET_ISSUE.value, description="Get details for a specific Jira issue by key", inputSchema={ "type": "object", "properties": { "issue_key": { "type": "string", "description": "The issue key (e.g., PROJECT-123)", } }, "required": ["issue_key"], }, ), Tool( name=JiraTools.SEARCH_ISSUES.value, description="Search for Jira issues using JQL (Jira Query Language)", inputSchema={ "type": "object", "properties": { "jql": { "type": "string", "description": "JQL query string (e.g., 'project = MYPROJ AND status = \"In Progress\"')", }, "max_results": { "type": "integer", "description": "Maximum number of results to return (default: 10)", }, }, "required": ["jql"], }, ), Tool( name=JiraTools.CREATE_ISSUE.value, description="Create a new Jira issue. Common issue types include 'Bug', 'Task', 'Story', 'Epic' (capitalization handled automatically)", inputSchema={ "type": "object", "properties": { "project": { "type": "string", "description": "Project key (e.g., 'MYPROJ')", }, "summary": { "type": "string", "description": "Issue summary/title", }, "description": { "type": "string", "description": "Issue description", }, "issue_type": { "type": "string", "description": "Issue type (e.g., 'Bug', 'Task', 'Story', 'Epic', 'New Feature', 'Improvement'). IMPORTANT: Types are case-sensitive and vary by Jira instance.", }, "fields": { "type": "object", "description": "Additional fields for the issue (optional)", }, }, "required": ["project", "summary", "description", "issue_type"], }, ), Tool( name=JiraTools.CREATE_ISSUES.value, description="Bulk create new Jira issues. IMPORTANT: For 'issue_type', use the exact case-sensitive types in your Jira instance (common: 'Bug', 'Task', 'Story', 'Epic')", inputSchema={ "type": "object", "properties": { "field_list": { "type": "array", "description": "A list of field dictionaries, each representing an issue to create", "items": { "type": "object", "description": "Field dictionary for a single issue", }, }, "prefetch": { "type": "boolean", "description": "Whether to reload created issues (default: true)", }, }, "required": ["field_list"], }, ), Tool( name=JiraTools.ADD_COMMENT.value, description="Add a comment to a Jira issue", inputSchema={ "type": "object", "properties": { "issue_key": { "type": "string", "description": "The issue key (e.g., PROJECT-123)", }, "comment": { "type": "string", "description": "The comment text", }, }, "required": ["issue_key", "comment"], }, ), Tool( name=JiraTools.GET_TRANSITIONS.value, description="Get available workflow transitions for a Jira issue", inputSchema={ "type": "object", "properties": { "issue_key": { "type": "string", "description": "The issue key (e.g., PROJECT-123)", } }, "required": ["issue_key"], }, ), Tool( name=JiraTools.TRANSITION_ISSUE.value, description="Transition a Jira issue to a new status", inputSchema={ "type": "object", "properties": { "issue_key": { "type": "string", "description": "The issue key (e.g., PROJECT-123)", }, "transition_id": { "type": "string", "description": "ID of the transition to perform (get IDs using get_transitions)", }, "comment": { "type": "string", "description": "Comment to add during transition (optional)", }, "fields": { "type": "object", "description": "Additional fields to update during transition (optional)", }, }, "required": ["issue_key", "transition_id"], }, ), Tool( name=JiraTools.GET_PROJECT_ISSUE_TYPES.value, description="Get all available issue types for a specific Jira project", inputSchema={ "type": "object", "properties": { "project_key": { "type": "string", "description": "The project key (e.g., 'MYPROJ')", } }, "required": ["project_key"], }, ), Tool( name=JiraTools.CREATE_PROJECT.value, description="Create a new Jira project using v3 REST API", inputSchema={ "type": "object", "properties": { "key": { "type": "string", "description": "Mandatory. Must match Jira project key requirements, usually only 2-10 uppercase characters.", }, "name": { "type": "string", "description": "If not specified it will use the key value.", }, "assignee": { "type": "string", "description": "Lead account ID or username (mapped to leadAccountId in v3 API).", }, "ptype": { "type": "string", "description": "Project type key: 'software', 'business', or 'service_desk'. Defaults to 'software'.", }, "template_name": { "type": "string", "description": "Project template key for creating from templates (mapped to projectTemplateKey in v3 API).", }, "avatarId": { "type": ["integer", "string"], "description": "ID of the avatar to use for the project.", }, "issueSecurityScheme": { "type": ["integer", "string"], "description": "Determines the security scheme to use.", }, "permissionScheme": { "type": ["integer", "string"], "description": "Determines the permission scheme to use.", }, "projectCategory": { "type": ["integer", "string"], "description": "Determines the category the project belongs to.", }, "notificationScheme": { "type": ["integer", "string"], "description": "Determines the notification scheme to use. Default is None.", }, "categoryId": { "type": ["integer", "string"], "description": "Same as projectCategory. Can be used interchangeably.", }, "url": { "type": "string", "description": "A link to information about the project, such as documentation.", }, }, "required": ["key"], }, ), ] @server.call_tool() async def call_tool( name: str, arguments: dict ) -> Sequence[TextContent | ImageContent | EmbeddedResource]: """Handle tool calls for Jira operations.""" logger.info(f"call_tool invoked. Tool: '{name}', Arguments: {arguments}") try: result: Any match name: case JiraTools.GET_PROJECTS.value: logger.info("About to AWAIT jira_server.get_jira_projects...") result = await jira_server.get_jira_projects() logger.info( f"COMPLETED await jira_server.get_jira_projects. Result has {len(result)} items." ) case JiraTools.GET_ISSUE.value: logger.info("Calling synchronous tool get_jira_issue...") issue_key = arguments.get("issue_key") if not issue_key: raise ValueError("Missing required argument: issue_key") result = jira_server.get_jira_issue(issue_key) logger.info("Synchronous tool get_jira_issue completed.") case JiraTools.SEARCH_ISSUES.value: logger.info("Calling async tool search_jira_issues...") jql = arguments.get("jql") if not jql: raise ValueError("Missing required argument: jql") max_results = arguments.get("max_results", 10) result = await jira_server.search_jira_issues(jql, max_results) logger.info("Async tool search_jira_issues completed.") case JiraTools.CREATE_ISSUE.value: logger.info("About to AWAIT jira_server.create_jira_issue...") required_args = ["project", "summary", "description", "issue_type"] if not all(arg in arguments for arg in required_args): missing = [arg for arg in required_args if arg not in arguments] raise ValueError( f"Missing required arguments: {', '.join(missing)}" ) result = await jira_server.create_jira_issue( arguments["project"], arguments["summary"], arguments["description"], arguments["issue_type"], arguments.get("fields", {}), ) logger.info("COMPLETED await jira_server.create_jira_issue.") case JiraTools.CREATE_ISSUES.value: logger.info("Calling async tool create_jira_issues...") field_list = arguments.get("field_list") if not field_list: raise ValueError("Missing required argument: field_list") prefetch = arguments.get("prefetch", True) result = await jira_server.create_jira_issues(field_list, prefetch) logger.info("Async tool create_jira_issues completed.") case JiraTools.ADD_COMMENT.value: logger.info("About to AWAIT jira_server.add_jira_comment...") issue_key = arguments.get("issue_key") comment_text = arguments.get("comment") or arguments.get("body") if not issue_key or not comment_text: raise ValueError( "Missing required arguments: issue_key and comment (or body)" ) result = await jira_server.add_jira_comment(issue_key, comment_text) logger.info("COMPLETED await jira_server.add_jira_comment.") case JiraTools.GET_TRANSITIONS.value: logger.info("About to AWAIT jira_server.get_jira_transitions...") issue_key = arguments.get("issue_key") if not issue_key: raise ValueError("Missing required argument: issue_key") result = await jira_server.get_jira_transitions(issue_key) logger.info("COMPLETED await jira_server.get_jira_transitions.") case JiraTools.TRANSITION_ISSUE.value: logger.info("Calling async tool transition_jira_issue...") issue_key = arguments.get("issue_key") transition_id = arguments.get("transition_id") if not issue_key or not transition_id: raise ValueError( "Missing required arguments: issue_key and transition_id" ) comment = arguments.get("comment") fields = arguments.get("fields") result = await jira_server.transition_jira_issue( issue_key, transition_id, comment, fields ) logger.info("Async tool transition_jira_issue completed.") case JiraTools.GET_PROJECT_ISSUE_TYPES.value: logger.info( "Calling asynchronous tool get_jira_project_issue_types..." ) project_key = arguments.get("project_key") if not project_key: raise ValueError("Missing required argument: project_key") result = await jira_server.get_jira_project_issue_types(project_key) logger.info( "Asynchronous tool get_jira_project_issue_types completed." ) case JiraTools.CREATE_PROJECT.value: logger.info("About to AWAIT jira_server.create_jira_project...") key = arguments.get("key") if not key: raise ValueError("Missing required argument: key") # Type conversion logic from original code for int_key in [ "avatarId", "issueSecurityScheme", "permissionScheme", "projectCategory", "notificationScheme", "categoryId", ]: if ( int_key in arguments and isinstance(arguments[int_key], str) and arguments[int_key].isdigit() ): arguments[int_key] = int(arguments[int_key]) result = await jira_server.create_jira_project(**arguments) logger.info("COMPLETED await jira_server.create_jira_project.") case _: raise ValueError(f"Unknown tool: {name}") logger.debug("Serializing result to JSON...") # Handle serialization properly for different result types if isinstance(result, list): # If it's a list, check each item individually serialized_result = [] for item in result: if hasattr(item, "model_dump"): serialized_result.append(item.model_dump()) else: # It's already a dict or basic type serialized_result.append(item) else: # Single item result if hasattr(result, "model_dump"): serialized_result = result.model_dump() else: # It's already a dict or basic type serialized_result = result json_result = json.dumps(serialized_result, indent=2) return [TextContent(type="text", text=json_result)] except Exception as e: logger.critical( f"FATAL error in call_tool for tool '{name}'", exc_info=True ) return [ TextContent( type="text", text=json.dumps( { "error": f"Error in tool '{name}': {type(e).__name__}: {str(e)}" } ), ) ] options = server.create_initialization_options() async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, options)