mcp_server.py•32.6 kB
#!/usr/bin/env python3
"""
MCP Server for Jira-GitLab Integration
Provides tools and resources for managing Jira issues and GitLab branches
"""
import asyncio
import json
import logging
import os
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import (
Resource,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
LoggingLevel
)
from connectors.jira_client import JiraClient
from connectors.github_client import GitHubClient
from connectors.openai_client import OpenAIClient, OpenAIError
from utils.error_handler import MCPError, handle_errors
from utils.config import load_config
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Initialize MCP server
server = Server("jira-gitlab-mcp")
# Global clients (will be initialized in main)
jira_client: Optional[JiraClient] = None
github_client: Optional[GitHubClient] = None
openai_client: Optional[OpenAIClient] = None
@server.list_resources()
async def handle_list_resources() -> List[Resource]:
"""List available MCP resources"""
return [
Resource(
uri="jira://issues",
name="Jira Issues",
description="Access to Jira issues in the configured project",
mimeType="application/json"
),
Resource(
uri="github://projects",
name="GitHub Projects",
description="Access to GitHub repositories and branches",
mimeType="application/json"
)
]
@server.read_resource()
async def handle_read_resource(uri: str) -> str:
"""Read MCP resource content"""
try:
if uri == "jira://issues":
if not jira_client:
raise MCPError("Jira client not initialized")
issues = await jira_client.get_issues()
return json.dumps({
"issues": [
{
"key": issue["key"],
"summary": issue["fields"]["summary"],
"status": issue["fields"]["status"]["name"],
"assignee": issue["fields"]["assignee"]["displayName"] if issue["fields"]["assignee"] else None,
"created": issue["fields"]["created"],
"updated": issue["fields"]["updated"]
}
for issue in issues
]
}, indent=2)
elif uri == "github://projects":
if not github_client:
raise MCPError("GitHub client not initialized")
projects = await github_client.get_projects()
return json.dumps({
"projects": [
{
"id": project["id"],
"name": project["name"],
"web_url": project["web_url"],
"default_branch": project["default_branch"]
}
for project in projects
]
}, indent=2)
else:
raise MCPError(f"Unknown resource URI: {uri}")
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
raise MCPError(f"Failed to read resource: {str(e)}")
@server.list_tools()
async def handle_list_tools() -> List[Tool]:
"""List available MCP tools"""
return [
Tool(
name="create_branch_for_issue",
description="Create a GitLab branch for a Jira issue and link them",
inputSchema={
"type": "object",
"properties": {
"issue_key": {
"type": "string",
"description": "Jira issue key (e.g., PROJ-123)"
},
"project_id": {
"type": "integer",
"description": "GitLab project ID"
},
"base_branch": {
"type": "string",
"description": "Base branch to create from (default: main)",
"default": "main"
}
},
"required": ["issue_key", "project_id"]
}
),
Tool(
name="get_jira_issues",
description="Fetch Jira issues using JQL query",
inputSchema={
"type": "object",
"properties": {
"jql": {
"type": "string",
"description": "JQL query string (optional)",
"default": f"project = {jira_client.project_key if jira_client else 'DEMO'} AND status = 'To Do'"
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 50
}
},
"required": []
}
),
Tool(
name="comment_on_issue",
description="Add a comment to a Jira issue",
inputSchema={
"type": "object",
"properties": {
"issue_key": {
"type": "string",
"description": "Jira issue key"
},
"comment": {
"type": "string",
"description": "Comment text to add"
}
},
"required": ["issue_key", "comment"]
}
),
Tool(
name="get_issues_by_tags",
description="Fetch Jira issues by project and tags (labels)",
inputSchema={
"type": "object",
"properties": {
"project_key": {
"type": "string",
"description": "Jira project key (e.g., PROJ, DEMO)"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "List of tags/labels to filter by (e.g., ['AI-Fix', 'AutoFix'])",
"minItems": 1,
"maxItems": 10
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 50,
"minimum": 1,
"maximum": 100
}
},
"required": ["project_key", "tags"]
}
),
Tool(
name="analyze_and_fix_issue",
description="Use AI to analyze a Jira issue and generate code fixes",
inputSchema={
"type": "object",
"properties": {
"issue_key": {
"type": "string",
"description": "Jira issue key to analyze and fix"
},
"model": {
"type": "string",
"description": "AI model to use for analysis",
"default": "gpt-4-turbo",
"enum": ["gpt-4-turbo", "gpt-4", "claude-3-sonnet", "claude-3-haiku"]
},
"validation_level": {
"type": "string",
"description": "Code validation strictness",
"default": "basic",
"enum": ["basic", "strict"]
},
"include_context": {
"type": "boolean",
"description": "Include repository context in analysis",
"default": true
}
},
"required": ["issue_key"]
}
),
Tool(
name="commit_ai_fix",
description="Commit AI-generated fixes to GitLab branch",
inputSchema={
"type": "object",
"properties": {
"project_id": {
"type": "integer",
"description": "GitLab project ID"
},
"branch_name": {
"type": "string",
"description": "Branch name to commit to"
},
"files": {
"type": "array",
"items": {
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"},
"action": {"type": "string", "enum": ["create", "update", "delete"], "default": "update"}
},
"required": ["path", "content"]
},
"description": "Files to commit with their content"
},
"commit_message": {
"type": "string",
"description": "Commit message"
}
},
"required": ["project_id", "branch_name", "files", "commit_message"]
}
),
Tool(
name="create_merge_request",
description="Create a GitLab merge request",
inputSchema={
"type": "object",
"properties": {
"project_id": {
"type": "integer",
"description": "GitLab project ID"
},
"source_branch": {
"type": "string",
"description": "Source branch name"
},
"target_branch": {
"type": "string",
"description": "Target branch name",
"default": "main"
},
"title": {
"type": "string",
"description": "Merge request title"
},
"description": {
"type": "string",
"description": "Merge request description"
},
"draft": {
"type": "boolean",
"description": "Create as draft MR",
"default": true
}
},
"required": ["project_id", "source_branch"]
}
),
Tool(
name="update_issue_status",
description="Update Jira issue status and add comments",
inputSchema={
"type": "object",
"properties": {
"issue_key": {
"type": "string",
"description": "Jira issue key"
},
"status": {
"type": "string",
"description": "New status (e.g., 'In Review', 'Done')"
},
"comment": {
"type": "string",
"description": "Optional comment to add with status change"
}
},
"required": ["issue_key", "status"]
}
),
Tool(
name="sre_ai_workflow",
description="Complete SRE AI workflow: fetch tagged issues, create fixes, and update status",
inputSchema={
"type": "object",
"properties": {
"project_key": {
"type": "string",
"description": "Jira project key"
},
"gitlab_project_id": {
"type": "integer",
"description": "GitLab project ID"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "Tags to filter issues by",
"default": ["AI-Fix", "AutoFix"]
},
"max_issues": {
"type": "integer",
"description": "Maximum issues to process",
"default": 5,
"minimum": 1,
"maximum": 10
},
"auto_merge": {
"type": "boolean",
"description": "Automatically merge approved fixes",
"default": false
}
},
"required": ["project_key", "gitlab_project_id"]
}
)
]
@server.call_tool()
@handle_errors
async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle MCP tool calls"""
if name == "create_branch_for_issue":
issue_key = arguments["issue_key"]
project_id = arguments["project_id"]
base_branch = arguments.get("base_branch", "main")
if not jira_client or not github_client:
raise MCPError("Clients not initialized")
# Create branch with correct naming convention: feature/{issueid}-fix
branch_name = f"feature/{issue_key}-fix"
try:
# Create the branch
branch_url = await github_client.create_branch(project_id, branch_name, base_branch)
# Add comment to Jira issue
comment = f"GitLab branch created: {branch_url}"
await jira_client.comment_issue(issue_key, comment)
return [TextContent(
type="text",
text=f"Successfully created branch '{branch_name}' and linked to issue {issue_key}.\nBranch URL: {branch_url}"
)]
except Exception as e:
logger.error(f"Error creating branch for issue {issue_key}: {e}")
raise MCPError(f"Failed to create branch: {str(e)}")
elif name == "get_jira_issues":
jql = arguments.get("jql", f"project = {jira_client.project_key if jira_client else 'DEMO'} AND status = 'To Do'")
max_results = arguments.get("max_results", 50)
if not jira_client:
raise MCPError("Jira client not initialized")
try:
issues = await jira_client.get_issues(jql, max_results)
result = {
"total": len(issues),
"issues": [
{
"key": issue["key"],
"summary": issue["fields"]["summary"],
"status": issue["fields"]["status"]["name"],
"assignee": issue["fields"]["assignee"]["displayName"] if issue["fields"]["assignee"] else None,
"priority": issue["fields"]["priority"]["name"] if issue["fields"]["priority"] else None,
"created": issue["fields"]["created"],
"updated": issue["fields"]["updated"]
}
for issue in issues
]
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error fetching Jira issues: {e}")
raise MCPError(f"Failed to fetch issues: {str(e)}")
elif name == "comment_on_issue":
issue_key = arguments["issue_key"]
comment = arguments["comment"]
if not jira_client:
raise MCPError("Jira client not initialized")
try:
await jira_client.comment_issue(issue_key, comment)
return [TextContent(
type="text",
text=f"Successfully added comment to issue {issue_key}"
)]
except Exception as e:
logger.error(f"Error commenting on issue {issue_key}: {e}")
raise MCPError(f"Failed to add comment: {str(e)}")
elif name == "get_issues_by_tags":
project_key = arguments["project_key"]
tags = arguments["tags"]
max_results = arguments.get("max_results", 50)
if not jira_client:
raise MCPError("Jira client not initialized")
try:
issues = await jira_client.get_issues_by_tags(project_key, tags, max_results)
result = {
"project": project_key,
"tags": tags,
"total": len(issues),
"issues": [
{
"key": issue["key"],
"summary": issue["fields"]["summary"],
"status": issue["fields"]["status"]["name"],
"assignee": issue["fields"]["assignee"]["displayName"] if issue["fields"]["assignee"] else None,
"priority": issue["fields"]["priority"]["name"] if issue["fields"]["priority"] else None,
"created": issue["fields"]["created"],
"updated": issue["fields"]["updated"]
}
for issue in issues
]
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error fetching issues by tags: {e}")
raise MCPError(f"Failed to fetch issues by tags: {str(e)}")
elif name == "analyze_and_fix_issue":
issue_key = arguments["issue_key"]
model = arguments.get("model", "gpt-4-turbo")
validation_level = arguments.get("validation_level", "basic")
include_context = arguments.get("include_context", True)
if not jira_client:
raise MCPError("Jira client not initialized")
try:
# Get issue details
issue = await jira_client.get_issue(issue_key)
if not issue:
raise MCPError(f"Issue {issue_key} not found")
# Use real OpenAI analysis if configured
if openai_client:
try:
analysis_result = await openai_client.analyze_issue(
issue, include_context, validation_level
)
# Format response
response = {
"issue_key": analysis_result.issue_key,
"summary": issue["fields"]["summary"],
"description": issue["fields"]["description"],
"analysis": analysis_result.analysis,
"suggested_fixes": analysis_result.suggested_fixes,
"confidence_score": analysis_result.confidence_score,
"model_used": analysis_result.model_used,
"tokens_used": analysis_result.tokens_used,
"cost_estimate": analysis_result.cost_estimate,
"validation_level": validation_level
}
return [TextContent(
type="text",
text=json.dumps(response, indent=2)
)]
except OpenAIError as e:
logger.warning(f"OpenAI analysis failed: {e}, falling back to simulation")
# Fall through to simulation
except Exception as e:
logger.warning(f"AI analysis error: {e}, falling back to simulation")
# Fall through to simulation
# Fallback: simulated analysis
analysis = {
"issue_key": issue_key,
"summary": issue["fields"]["summary"],
"description": issue["fields"]["description"],
"analysis": f"Simulated AI analysis (OpenAI not configured or failed)",
"suggested_fixes": [
{
"path": "src/main.py",
"content": f"# AI-generated fix for {issue_key}\n# {issue['fields']['summary']}\nprint('Fixed issue')",
"action": "update",
"language": "python",
"description": "Example fix implementation"
}
],
"confidence_score": 0.5,
"validation_level": validation_level,
"model_used": "simulation",
"tokens_used": 0,
"cost_estimate": 0.0
}
return [TextContent(
type="text",
text=json.dumps(analysis, indent=2)
)]
except Exception as e:
logger.error(f"Error analyzing issue {issue_key}: {e}")
raise MCPError(f"Failed to analyze issue: {str(e)}")
elif name == "commit_ai_fix":
project_id = arguments["project_id"]
branch_name = arguments["branch_name"]
files = arguments["files"]
commit_message = arguments["commit_message"]
if not github_client:
raise MCPError("GitHub client not initialized")
try:
# Validate code changes first
from utils.code_validator import validate_code_changes
validation_results = await validate_code_changes(files)
# Check for validation errors
errors = []
for path, result in validation_results.items():
if not result.valid:
errors.extend([f"{path}: {error}" for error in result.errors])
if errors:
raise MCPError(f"Code validation failed: {'; '.join(errors)}")
# Commit files
commit_url = await github_client.commit_files(project_id, branch_name, files, commit_message)
return [TextContent(
type="text",
text=f"Successfully committed {len(files)} files to branch '{branch_name}'.\nCommit URL: {commit_url}"
)]
except Exception as e:
logger.error(f"Error committing AI fix: {e}")
raise MCPError(f"Failed to commit AI fix: {str(e)}")
elif name == "create_merge_request":
project_id = arguments["project_id"]
source_branch = arguments["source_branch"]
target_branch = arguments.get("target_branch", "main")
title = arguments.get("title", "")
description = arguments.get("description", "")
draft = arguments.get("draft", True)
if not github_client:
raise MCPError("GitHub client not initialized")
try:
mr_url = await github_client.create_merge_request(
project_id, source_branch, target_branch, title, description, draft
)
return [TextContent(
type="text",
text=f"Successfully created merge request from '{source_branch}' to '{target_branch}'.\nMR URL: {mr_url}"
)]
except Exception as e:
logger.error(f"Error creating merge request: {e}")
raise MCPError(f"Failed to create merge request: {str(e)}")
elif name == "update_issue_status":
issue_key = arguments["issue_key"]
status = arguments["status"]
comment = arguments.get("comment")
if not jira_client:
raise MCPError("Jira client not initialized")
try:
await jira_client.update_issue_status(issue_key, status, comment)
return [TextContent(
type="text",
text=f"Successfully updated issue {issue_key} status to '{status}'"
)]
except Exception as e:
logger.error(f"Error updating issue status: {e}")
raise MCPError(f"Failed to update issue status: {str(e)}")
elif name == "sre_ai_workflow":
project_key = arguments["project_key"]
gitlab_project_id = arguments["gitlab_project_id"]
tags = arguments.get("tags", ["AI-Fix", "AutoFix"])
max_issues = arguments.get("max_issues", 5)
auto_merge = arguments.get("auto_merge", False)
if not jira_client or not github_client:
raise MCPError("Clients not initialized")
try:
# Step 1: Fetch tagged issues
issues = await jira_client.get_issues_by_tags(project_key, tags, max_issues)
if not issues:
return [TextContent(
type="text",
text=f"No issues found with tags {tags} in project {project_key}"
)]
workflow_results = []
for issue in issues:
issue_key = issue["key"]
try:
# Step 2: Create branch
branch_name = f"ai-fix/{issue_key}"
branch_url = await github_client.create_branch(gitlab_project_id, branch_name, "main")
# Step 3: Simulate AI analysis and fix generation
# In real implementation, this would call an LLM
ai_fix = {
"path": f"fix_{issue_key.lower()}.py",
"content": f"# AI-generated fix for {issue_key}\n# {issue['fields']['summary']}\nprint('Fixed issue {issue_key}')",
"action": "create"
}
# Step 4: Commit fix
commit_message = f"AI-generated fix for {issue_key}: {issue['fields']['summary']}"
commit_url = await github_client.commit_files(
gitlab_project_id, branch_name, [ai_fix], commit_message
)
# Step 5: Create MR
mr_title = f"AI Fix: {issue['fields']['summary']}"
mr_description = f"Automated fix for issue {issue_key}\n\nOriginal issue: {issue['fields']['summary']}"
mr_url = await github_client.create_merge_request(
gitlab_project_id, branch_name, "main", mr_title, mr_description, draft=True
)
# Step 6: Update Jira status
status_comment = f"AI-generated fix created:\n- Branch: {branch_url}\n- Commit: {commit_url}\n- MR: {mr_url}"
await jira_client.update_issue_status(issue_key, "In Review", status_comment)
workflow_results.append({
"issue_key": issue_key,
"status": "success",
"branch_url": branch_url,
"commit_url": commit_url,
"mr_url": mr_url
})
except Exception as e:
logger.error(f"Error processing issue {issue_key}: {e}")
workflow_results.append({
"issue_key": issue_key,
"status": "error",
"error": str(e)
})
result = {
"project_key": project_key,
"gitlab_project_id": gitlab_project_id,
"tags": tags,
"processed_issues": len(workflow_results),
"successful": len([r for r in workflow_results if r["status"] == "success"]),
"failed": len([r for r in workflow_results if r["status"] == "error"]),
"results": workflow_results
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error in SRE AI workflow: {e}")
raise MCPError(f"Failed to execute SRE AI workflow: {str(e)}")
else:
raise MCPError(f"Unknown tool: {name}")
async def start_stdio_server():
"""Start the stdio-based MCP server"""
global jira_client, github_client, openai_client
try:
# Load configuration
config = load_config()
# Initialize clients
jira_client = JiraClient(config["jira"])
github_client = GitHubClient(config["github"])
# Initialize OpenAI client if configured
if "ai" in config and config["ai"].get("provider") == "openai":
try:
openai_client = OpenAIClient(config["ai"])
await openai_client.test_connection()
logger.info("OpenAI client initialized successfully")
except Exception as e:
logger.warning(f"Failed to initialize OpenAI client: {e}")
logger.info("Continuing without AI capabilities")
openai_client = None
else:
logger.info("OpenAI not configured, AI analysis will use simulation")
openai_client = None
# Test connections
await jira_client.test_connection()
await github_client.test_connection()
logger.info("MCP Server initialized successfully")
# Run the server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="jira-github-mcp",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(
prompts_changed=False,
resources_changed=True,
tools_changed=True
),
experimental_capabilities={}
),
),
)
except Exception as e:
logger.error(f"Failed to start MCP server: {e}")
raise
async def start_websocket_server():
"""Start the WebSocket-based MCP server"""
try:
# Import WebSocket server
from websocket_server import WebSocketMCPServer
# Load configuration
config = load_config()
# Create and start WebSocket server
ws_server = WebSocketMCPServer(config)
await ws_server.start_server()
except Exception as e:
logger.error(f"Failed to start WebSocket server: {e}")
raise
async def main():
"""Main server entry point"""
try:
# Load configuration
config = load_config()
# Check server mode
server_mode = config.get("server", {}).get("mode", "stdio")
if server_mode == "websocket":
logger.info("Starting in WebSocket mode")
await start_websocket_server()
else:
logger.info("Starting in stdio mode")
await start_stdio_server()
except KeyboardInterrupt:
logger.info("Server shutdown requested")
except Exception as e:
logger.error(f"Failed to start server: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())