Python Jira MCP Server

src
tools
"""
Jira tools for Model Context Protocol server.
This module defines tools for interacting with Jira API.
"""
import os
import base64
import json
from typing import List, Optional, Dict, Any, Union
import aiohttp
from pydantic import BaseModel, Field


class JQLSearchInput(BaseModel):
    """Input schema for JQL search tool."""
    jql: str = Field(..., description="JQL query string")
    next_page_token: Optional[str] = Field(None, description="Token for pagination")
    max_results: Optional[int] = Field(None, description="Maximum number of results to return")
    fields: Optional[List[str]] = Field(None, description="Fields to include in the response")
    expand: Optional[str] = Field(None, description="Additional data to expand")


class GetIssueInput(BaseModel):
    """Input schema for get_issue tool."""
    issue_id_or_key: str = Field(..., description="Issue identifier (ID or key)")
    fields: Optional[List[str]] = Field(None, description="Fields to return")
    expand: Optional[str] = Field(None, description="Additional data to expand")
    properties: Optional[List[str]] = Field(None, description="Properties to include")
    fail_fast: Optional[bool] = Field(None, description="Fail quickly on errors")


class JiraClient:
    """Client for Jira API."""
    
    def __init__(self):
        self.base_url = os.environ.get("JIRA_URL")
        self.email = os.environ.get("JIRA_EMAIL")
        self.api_token = os.environ.get("JIRA_API_TOKEN")
        
        if not all([self.base_url, self.email, self.api_token]):
            raise ValueError("Missing Jira configuration. Check environment variables.")
    
    def get_auth_header(self) -> Dict[str, str]:
        """Get authorization header for Jira API."""
        auth_str = f"{self.email}:{self.api_token}"
        auth_bytes = auth_str.encode("utf-8")
        auth_b64 = base64.b64encode(auth_bytes).decode("utf-8")
        return {"Authorization": f"Basic {auth_b64}"}
    
    async def search_issues(self, params: JQLSearchInput) -> Dict[str, Any]:
        """Search Jira issues using JQL."""
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
            **self.get_auth_header()
        }
        
        query_params = {"jql": params.jql}
        
        if params.max_results:
            query_params["maxResults"] = params.max_results
        
        if params.fields:
            query_params["fields"] = ",".join(params.fields)
        
        if params.expand:
            query_params["expand"] = params.expand
        
        if params.next_page_token:
            query_params["startAt"] = params.next_page_token
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.base_url}/rest/api/3/search",
                headers=headers,
                params=query_params
            ) as response:
                if response.status != 200:
                    error_text = await response.text()
                    return {
                        "isError": True,
                        "content": f"Jira API error: {response.status} - {error_text}"
                    }
                
                data = await response.json()
                return {
                    "isError": False,
                    "content": data
                }
    
    async def get_issue(self, params: GetIssueInput) -> Dict[str, Any]:
        """Get Jira issue details by ID or key."""
        headers = {
            "Accept": "application/json",
            **self.get_auth_header()
        }
        
        query_params = {}
        
        if params.fields:
            query_params["fields"] = ",".join(params.fields)
        
        if params.expand:
            query_params["expand"] = params.expand
        
        if params.properties:
            query_params["properties"] = ",".join(params.properties)
        
        if params.fail_fast is not None:
            query_params["failFast"] = str(params.fail_fast).lower()
        
        async with aiohttp.ClientSession() as session:
            async with session.get(
                f"{self.base_url}/rest/api/3/issue/{params.issue_id_or_key}",
                headers=headers,
                params=query_params
            ) as response:
                if response.status != 200:
                    error_text = await response.text()
                    return {
                        "isError": True,
                        "content": f"Jira API error: {response.status} - {error_text}"
                    }
                
                data = await response.json()
                return {
                    "isError": False,
                    "content": data
                }


# Tool handler functions
async def jql_search(params: Dict[str, Any]) -> Dict[str, Any]:
    """Perform JQL search in Jira."""
    try:
        input_params = JQLSearchInput(**params)
        client = JiraClient()
        return await client.search_issues(input_params)
    except Exception as e:
        return {
            "isError": True,
            "content": f"Error in jql_search: {str(e)}"
        }


async def get_issue(params: Dict[str, Any]) -> Dict[str, Any]:
    """Get Jira issue details."""
    try:
        input_params = GetIssueInput(**params)
        client = JiraClient()
        return await client.get_issue(input_params)
    except Exception as e:
        return {
            "isError": True,
            "content": f"Error in get_issue: {str(e)}"
        } 