"""Simplified MCP server implementation for JIRA using FastMCP."""
import logging
from typing import Any, Dict, List, Optional
from fastmcp import FastMCP
from mcp_jira.client import JiraClient
from mcp_jira.config import Config, get_config
from mcp_jira.operations import (
AttachmentOperations,
CommentOperations,
IssueOperations,
ProjectOperations,
SearchOperations,
TransitionOperations,
)
logger = logging.getLogger(__name__)
# Global instances
_jira_client: Optional[JiraClient] = None
_config: Optional[Config] = None
def get_jira_client() -> JiraClient:
"""Get or create the global JIRA client instance."""
global _jira_client, _config
if _jira_client is None:
_config = get_config()
_jira_client = JiraClient(_config)
# Test connection
connection_info = _jira_client.test_connection()
if not connection_info.get("connected"):
raise RuntimeError(f"Failed to connect to JIRA: {connection_info.get('error')}")
logger.info(
f"Connected to JIRA: {connection_info.get('server_title')} "
f"v{connection_info.get('server_version')}, "
f"user: {connection_info.get('current_user')}"
)
return _jira_client
# Initialize FastMCP server
mcp = FastMCP("JIRA MCP Server")
# ==================== TOOLS ====================
@mcp.tool()
def jira_create_issue(
project: str,
summary: str,
issue_type: str,
description: Optional[str] = None,
priority: Optional[str] = None,
assignee: Optional[str] = None,
labels: Optional[List[str]] = None,
components: Optional[List[str]] = None,
custom_fields: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Create a new JIRA issue with support for custom fields.
Args:
project: Project key
summary: Issue summary
issue_type: Issue type (e.g., Bug, Task, Story)
description: Issue description (optional)
priority: Priority name (optional)
assignee: Assignee account ID or username (optional)
labels: List of labels (optional)
components: List of component names (optional)
custom_fields: Custom fields as key-value pairs (optional)
Returns:
Dictionary with created issue information
"""
client = get_jira_client()
ops = IssueOperations(client)
return ops.create_issue(
project=project,
summary=summary,
issue_type=issue_type,
description=description,
priority=priority,
assignee=assignee,
labels=labels,
components=components,
custom_fields=custom_fields,
)
@mcp.tool()
def jira_get_issue(
issue_key: str,
fields: Optional[List[str]] = None,
expand: Optional[str] = None,
) -> Dict[str, Any]:
"""Get details of a JIRA issue including custom fields.
Args:
issue_key: Issue key (e.g., PROJ-123)
fields: List of fields to retrieve (optional)
expand: Comma-separated entities to expand (optional)
Returns:
Dictionary with issue details
"""
client = get_jira_client()
ops = IssueOperations(client)
return ops.get_issue(issue_key=issue_key, fields=fields, expand=expand)
@mcp.tool()
def jira_update_issue(
issue_key: str,
fields: Optional[Dict[str, Any]] = None,
custom_fields: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Update an existing JIRA issue including custom fields.
Args:
issue_key: Issue key
fields: Standard fields to update (optional)
custom_fields: Custom fields to update (optional)
Returns:
Dictionary with update result
"""
client = get_jira_client()
ops = IssueOperations(client)
return ops.update_issue(issue_key=issue_key, fields=fields, custom_fields=custom_fields)
@mcp.tool()
def jira_delete_issue(issue_key: str) -> Dict[str, Any]:
"""Delete a JIRA issue.
Args:
issue_key: Issue key
Returns:
Dictionary with deletion result
"""
client = get_jira_client()
ops = IssueOperations(client)
return ops.delete_issue(issue_key=issue_key)
@mcp.tool()
def jira_assign_issue(issue_key: str, assignee: Optional[str] = None) -> Dict[str, Any]:
"""Assign a JIRA issue to a user or unassign.
Args:
issue_key: Issue key
assignee: Account ID, username, or None to unassign
Returns:
Dictionary with assignment result
"""
client = get_jira_client()
ops = IssueOperations(client)
return ops.assign_issue(issue_key=issue_key, assignee=assignee)
@mcp.tool()
def jira_search_issues(
jql: str,
start_at: int = 0,
max_results: int = 50,
fields: Optional[List[str]] = None,
) -> Dict[str, Any]:
"""Search for issues using JQL (JIRA Query Language).
Args:
jql: JQL query string
start_at: Starting index for pagination (default: 0)
max_results: Maximum number of results (default: 50)
fields: List of fields to retrieve (optional)
Returns:
Dictionary with search results
"""
client = get_jira_client()
ops = SearchOperations(client)
return ops.search_issues(jql=jql, start_at=start_at, max_results=max_results, fields=fields)
@mcp.tool()
def jira_add_comment(issue_key: str, body: str) -> Dict[str, Any]:
"""Add a comment to a JIRA issue.
Args:
issue_key: Issue key
body: Comment text
Returns:
Dictionary with comment information
"""
client = get_jira_client()
ops = CommentOperations(client)
return ops.add_comment(issue_key=issue_key, body=body)
@mcp.tool()
def jira_transition_issue(
issue_key: str,
transition: str,
fields: Optional[Dict[str, Any]] = None,
custom_fields: Optional[Dict[str, Any]] = None,
comment: Optional[str] = None,
) -> Dict[str, Any]:
"""Transition a JIRA issue through workflow (e.g., move to In Progress, Done).
Args:
issue_key: Issue key
transition: Transition name or ID
fields: Fields to set during transition (optional)
custom_fields: Custom fields to set (optional)
comment: Comment to add with transition (optional)
Returns:
Dictionary with transition result
"""
client = get_jira_client()
ops = TransitionOperations(client)
return ops.transition_issue(
issue_key=issue_key,
transition=transition,
fields=fields,
custom_fields=custom_fields,
comment=comment,
)
@mcp.tool()
def jira_get_transitions(issue_key: str) -> List[Dict[str, Any]]:
"""Get available workflow transitions for an issue.
Args:
issue_key: Issue key
Returns:
List of available transitions
"""
client = get_jira_client()
ops = TransitionOperations(client)
return ops.get_transitions(issue_key=issue_key)
@mcp.tool()
def jira_upload_attachment(issue_key: str, attachment_path: str) -> Dict[str, Any]:
"""Upload an attachment to a JIRA issue.
Args:
issue_key: Issue key
attachment_path: Path to file to attach
Returns:
Dictionary with attachment information
"""
client = get_jira_client()
ops = AttachmentOperations(client)
return ops.add_attachment(issue_key=issue_key, attachment=attachment_path)
@mcp.tool()
def jira_list_projects() -> List[Dict[str, Any]]:
"""List all accessible JIRA projects.
Returns:
List of projects
"""
client = get_jira_client()
ops = ProjectOperations(client)
return ops.list_projects()
@mcp.tool()
def jira_get_custom_fields() -> List[Dict[str, Any]]:
"""Get all custom field definitions for JIRA.
Returns:
List of custom fields with their metadata
"""
client = get_jira_client()
custom_fields = client.custom_fields.get_all_custom_fields()
return [
{
"id": cf.id,
"name": cf.name,
"custom": cf.custom,
"searchable": cf.searchable,
"schema": cf.schema,
}
for cf in custom_fields
]
# ==================== RESOURCES ====================
@mcp.resource("jira://projects")
def get_projects() -> str:
"""Get list of all accessible JIRA projects."""
import json
client = get_jira_client()
ops = ProjectOperations(client)
projects = ops.list_projects()
return json.dumps(projects, indent=2, default=str)
@mcp.resource("jira://custom-fields")
def get_custom_fields_resource() -> str:
"""Get all custom field definitions."""
import json
client = get_jira_client()
custom_fields = client.custom_fields.get_all_custom_fields()
fields_data = [
{
"id": cf.id,
"name": cf.name,
"custom": cf.custom,
"searchable": cf.searchable,
"schema": cf.schema,
}
for cf in custom_fields
]
return json.dumps(fields_data, indent=2, default=str)
@mcp.resource("jira://current-user")
def get_current_user() -> str:
"""Get information about the authenticated user."""
import json
client = get_jira_client()
jira = client.jira
user = jira.current_user()
server_info = jira.server_info()
return json.dumps(
{
"user": user,
"server": server_info.get("serverTitle"),
"version": server_info.get("version"),
},
indent=2,
default=str,
)