""" Jira Client for MCP Server Integration """ import os import logging from jira import JIRA logger = logging.getLogger(__name__) class JiraClient: """Jira client for MCP server tools""" def __init__(self): self.jira = None self.connected = False self._initialize_client() def _initialize_client(self): """Initialize Jira client with environment variables""" try: jira_url = os.getenv("JIRA_URL") if not jira_url: logger.warning("JIRA_URL not configured") return # Try Personal Access Token first (recommended) personal_token = os.getenv("JIRA_PERSONAL_TOKEN") if personal_token: try: self.jira = JIRA(server=jira_url, token_auth=personal_token) # Test connection self.jira.server_info() self.connected = True logger.info("Connected to Jira with Personal Token") return except Exception as e: logger.error(f"Failed to connect to Jira with Personal Token: {e}") # Fallback to username/API token username = os.getenv("JIRA_USERNAME") api_token = os.getenv("JIRA_API_TOKEN") if username and api_token: try: self.jira = JIRA(server=jira_url, basic_auth=(username, api_token)) # Test connection self.jira.server_info() self.connected = True logger.info("Connected to Jira with Username/API Token") return except Exception as e: logger.error( f"Failed to connect to Jira with Username/API Token: {e}" ) logger.warning("No valid Jira credentials found") except Exception as e: logger.error(f"Error initializing Jira client: {e}") self.connected = False def is_connected(self): """Check if Jira client is connected""" return self.connected and self.jira is not None def list_projects(self): """List all accessible Jira projects""" if not self.is_connected(): return {"error": "Not connected to Jira"} try: projects = self.jira.projects() project_list = [] for project in projects: project_list.append( { "key": project.key, "name": project.name, "project_type": getattr(project, "projectTypeKey", "unknown"), "lead": ( getattr(project.lead, "displayName", "unknown") if hasattr(project, "lead") else "unknown" ), } ) return {"projects": project_list, "count": len(project_list)} except Exception as e: logger.error(f"Error listing projects: {e}") return {"error": str(e)} def search_issues(self, jql, max_results=100): """Search Jira issues using JQL""" if not self.is_connected(): return {"error": "Not connected to Jira"} try: issues = self.jira.search_issues(jql, maxResults=max_results) issue_list = [] for issue in issues: issue_list.append( { "key": issue.key, "summary": issue.fields.summary, "status": issue.fields.status.name, "issue_type": issue.fields.issuetype.name, "priority": ( issue.fields.priority.name if issue.fields.priority else "None" ), "assignee": ( issue.fields.assignee.displayName if issue.fields.assignee else "Unassigned" ), "reporter": ( issue.fields.reporter.displayName if issue.fields.reporter else "Unknown" ), "created": str(issue.fields.created), "updated": str(issue.fields.updated), } ) return { "issues": issue_list, "count": len(issue_list), "total": issues.total, } except Exception as e: logger.error(f"Error searching issues: {e}") return {"error": str(e)} def get_issue(self, issue_key): """Get detailed information about a specific issue""" if not self.is_connected(): return {"error": "Not connected to Jira"} try: issue = self.jira.issue(issue_key) return { "key": issue.key, "summary": issue.fields.summary, "description": issue.fields.description or "No description", "status": issue.fields.status.name, "issue_type": issue.fields.issuetype.name, "priority": ( issue.fields.priority.name if issue.fields.priority else "None" ), "assignee": ( issue.fields.assignee.displayName if issue.fields.assignee else "Unassigned" ), "reporter": ( issue.fields.reporter.displayName if issue.fields.reporter else "Unknown" ), "created": str(issue.fields.created), "updated": str(issue.fields.updated), "project": { "key": issue.fields.project.key, "name": issue.fields.project.name, }, "labels": issue.fields.labels, "components": ( [comp.name for comp in issue.fields.components] if issue.fields.components else [] ), } except Exception as e: logger.error(f"Error getting issue {issue_key}: {e}") return {"error": str(e)} def get_issue_comments(self, issue_key): """Get comments for a specific issue""" if not self.is_connected(): return {"error": "Not connected to Jira"} try: issue = self.jira.issue(issue_key) comments = self.jira.comments(issue) comment_list = [] for comment in comments: comment_list.append( { "id": comment.id, "author": comment.author.displayName, "body": comment.body, "created": str(comment.created), "updated": str(comment.updated), } ) return { "issue_key": issue_key, "comments": comment_list, "count": len(comment_list), } except Exception as e: logger.error(f"Error getting comments for {issue_key}: {e}") return {"error": str(e)} def create_issue( self, project_key, summary, description, issue_type="Task", priority="Medium", ): """Create a new Jira issue""" if not self.is_connected(): return {"error": "Not connected to Jira"} try: issue_dict = { "project": {"key": project_key}, "summary": summary, "description": description, "issuetype": {"name": issue_type}, } # Add priority if available try: priorities = self.jira.priorities() priority_names = [p.name for p in priorities] if priority in priority_names: issue_dict["priority"] = {"name": priority} except Exception: logger.warning(f"Could not set priority {priority}") new_issue = self.jira.create_issue(fields=issue_dict) return { "key": new_issue.key, "summary": summary, "status": "Created", "url": f"{self.jira.server_url}/browse/{new_issue.key}", } except Exception as e: logger.error(f"Error creating issue: {e}") return {"error": str(e)}