MCP JIRA Server

"""JIRA client with authentication and custom fields support.""" import logging from typing import Any, Dict, Optional from jira import JIRA from jira.exceptions import JIRAError from requests import Session from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry from mcp_jira.auth import AuthManager from mcp_jira.client.custom_fields import CustomFieldsManager from mcp_jira.config import Config logger = logging.getLogger(__name__) class JiraClient: """Enhanced JIRA client with authentication and custom fields support.""" def __init__(self, config: Config): """Initialize JIRA client. Args: config: Application configuration """ self.config = config self.auth_manager = AuthManager(config) self._jira: Optional[JIRA] = None self._custom_fields_manager: Optional[CustomFieldsManager] = None def _create_session(self) -> Session: """Create requests session with retry logic. Returns: Configured requests Session """ session = Session() # Configure retries retry_strategy = Retry( total=self.config.max_retries, backoff_factor=self.config.retry_backoff_factor, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE", "POST"], ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) # Set authentication session.auth = self.auth_manager.get_auth() return session def connect(self) -> JIRA: """Connect to JIRA and return authenticated client. Returns: Authenticated JIRA client instance Raises: JIRAError: If connection fails """ if self._jira is not None: return self._jira try: # Create session with authentication session = self._create_session() # Initialize JIRA client options = { "server": self.config.jira_url, "rest_api_version": self.config.jira_api_version, "timeout": self.config.request_timeout, } logger.info(f"Connecting to JIRA at {self.config.jira_url}") self._jira = JIRA(options=options, session=session) # Test connection server_info = self._jira.server_info() logger.info( f"Successfully connected to JIRA {server_info.get('version', 'unknown')}" ) # Initialize custom fields manager self._custom_fields_manager = CustomFieldsManager(self._jira) return self._jira except JIRAError as e: logger.error(f"Failed to connect to JIRA: {e}") raise except Exception as e: logger.error(f"Unexpected error connecting to JIRA: {e}") raise @property def jira(self) -> JIRA: """Get JIRA client instance, connecting if necessary. Returns: Authenticated JIRA client """ if self._jira is None: self.connect() return self._jira # type: ignore @property def custom_fields(self) -> CustomFieldsManager: """Get custom fields manager. Returns: CustomFieldsManager instance """ if self._custom_fields_manager is None: # Ensure connection is established self.connect() return self._custom_fields_manager # type: ignore def reconnect(self) -> JIRA: """Reconnect to JIRA (useful after credential refresh). Returns: New authenticated JIRA client instance """ logger.info("Reconnecting to JIRA") self._jira = None self._custom_fields_manager = None self.auth_manager.refresh_credentials() return self.connect() def test_connection(self) -> Dict[str, Any]: """Test JIRA connection and return server info. Returns: Dictionary with server information Raises: JIRAError: If connection test fails """ try: jira = self.jira server_info = jira.server_info() current_user = jira.current_user() return { "connected": True, "server_version": server_info.get("version"), "server_title": server_info.get("serverTitle"), "base_url": server_info.get("baseUrl"), "current_user": current_user, } except JIRAError as e: logger.error(f"Connection test failed: {e}") return { "connected": False, "error": str(e), }