Glama
Python Jira MCP Server

"""
Jira MCP Server implementation.
This module provides a Model Context Protocol server for Jira API integration.
"""
import os
import sys
import asyncio
import json
import logging
from typing import Dict, Any, Callable, Awaitable, Optional, List

# Try to import from the official MCP Python SDK
try:
    from mcp.stdio_transport import StdioTransport
    from mcp.server import MCPServer
    HAS_MCP_SDK = True
except ImportError:
    # Fallback to a minimal implementation if the SDK is not available
    HAS_MCP_SDK = False

# Import our tool implementations
from src.tools import jql_search, get_issue
from src.tool_schemas import get_tool_schemas

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    filename="jira_mcp.log"
)
logger = logging.getLogger("jira_mcp")


class JiraMCPServer:
    """MCP Server implementation for Jira API integration."""
    
    def __init__(self):
        """Initialize the Jira MCP server."""
        # Map tool names to their handler functions
        self.tools = {
            "jql_search": jql_search,
            "get_issue": get_issue,
        }
        
        # Load environment variables from .env file if python-dotenv is available
        try:
            from dotenv import load_dotenv
            load_dotenv()
            logger.info("Loaded environment from .env file")
        except ImportError:
            logger.warning("python-dotenv not found, skipping .env loading")
    
    async def handle_tool_call(self, tool_name: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """Handle a tool call with the given name and parameters."""
        logger.info(f"Handling tool call: {tool_name}")
        
        if tool_name not in self.tools:
            logger.error(f"Tool not found: {tool_name}")
            return {
                "isError": True,
                "content": f"Tool not found: {tool_name}"
            }
        
        try:
            # Call the appropriate tool handler
            tool_handler = self.tools[tool_name]
            result = await tool_handler(params)
            logger.info(f"Tool call completed: {tool_name}")
            return result
        except Exception as e:
            logger.exception(f"Error in tool {tool_name}: {e}")
            return {
                "isError": True,
                "content": f"Error in {tool_name}: {str(e)}"
            }
    
    async def start_with_sdk(self):
        """Start the MCP server using the official MCP SDK."""
        transport = StdioTransport()
        server = MCPServer(transport)
        
        # Register tools with the server
        for schema in get_tool_schemas():
            name = schema["name"]
            
            async def handler(params, tool_name=name):
                return await self.handle_tool_call(tool_name, params)
            
            server.register_tool(schema, handler)
        
        logger.info("Starting Jira MCP server with official SDK")
        await server.serve()
    
    async def start_minimal(self):
        """Start a minimal MCP server implementation without the SDK."""
        logger.info("Starting minimal Jira MCP server (SDK not available)")
        
        # Read from stdin and write to stdout
        while True:
            try:
                # Read a line from stdin
                line = await asyncio.to_thread(sys.stdin.readline)
                if not line:
                    break
                
                # Parse the JSON message
                message = json.loads(line)
                if "type" not in message:
                    continue
                
                if message["type"] == "tool_call":
                    # Handle tool call
                    tool_name = message.get("name")
                    params = message.get("parameters", {})
                    
                    result = await self.handle_tool_call(tool_name, params)
                    
                    # Send the response
                    response = {
                        "type": "tool_call_result",
                        "id": message.get("id"),
                        "result": result
                    }
                    print(json.dumps(response), flush=True)
                
                elif message["type"] == "discover":
                    # Handle tool discovery
                    response = {
                        "type": "discover_response",
                        "tools": get_tool_schemas()
                    }
                    print(json.dumps(response), flush=True)
            
            except json.JSONDecodeError:
                logger.error("Failed to parse JSON message")
            except Exception as e:
                logger.exception(f"Error processing message: {e}")
    
    async def start(self):
        """Start the Jira MCP server."""
        if HAS_MCP_SDK:
            await self.start_with_sdk()
        else:
            await self.start_minimal()


async def main():
    """Main entry point for the Jira MCP server."""
    server = JiraMCPServer()
    await server.start()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Server stopped by user")
    except Exception as e:
        logger.exception(f"Unhandled error: {e}")
        sys.exit(1) 