#!/usr/bin/env python3
"""
Jira MCP Server
A Model Context Protocol server that integrates with Jira Cloud to provide
issue management, project tracking, and team collaboration capabilities.
Features:
- Search and manage issues (Stories, Bugs, Tasks)
- Create and update tickets
- Project and board management
- User and assignment management
- Comment management
"""
import os
import logging
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin
import httpx
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Initialize FastMCP server
mcp = FastMCP("jira")
# Configure logging to stderr (not stdout for MCP)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
)
logger = logging.getLogger(__name__)
# Jira configuration
JIRA_BASE_URL = os.getenv("JIRA_BASE_URL") # e.g., "https://yourcompany.atlassian.net"
JIRA_EMAIL = os.getenv("JIRA_EMAIL") # Your Jira account email
JIRA_API_TOKEN = os.getenv("JIRA_API_TOKEN") # Your Jira API token
# Validate required environment variables
if not all([JIRA_BASE_URL, JIRA_EMAIL, JIRA_API_TOKEN]):
logger.error("Missing required environment variables. Please set JIRA_BASE_URL, JIRA_EMAIL, and JIRA_API_TOKEN")
class JiraClient:
"""Jira API client with authentication and error handling."""
def __init__(self):
self.base_url = JIRA_BASE_URL
self.auth = (JIRA_EMAIL, JIRA_API_TOKEN)
self.headers = {
"Accept": "application/json",
"Content-Type": "application/json"
}
async def make_request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make authenticated request to Jira API."""
url = urljoin(f"{self.base_url}/rest/api/3/", endpoint)
async with httpx.AsyncClient() as client:
try:
response = await client.request(
method=method,
url=url,
auth=self.auth,
headers=self.headers,
timeout=30.0,
**kwargs
)
response.raise_for_status()
return response.json() if response.content else {}
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error {e.response.status_code}: {e.response.text}")
raise Exception(f"Jira API error: {e.response.status_code} - {e.response.text}")
except Exception as e:
logger.error(f"Request failed: {str(e)}")
raise Exception(f"Failed to connect to Jira: {str(e)}")
# Initialize Jira client
jira_client = JiraClient()
def format_issue(issue: Dict[str, Any]) -> str:
"""Format a Jira issue for display."""
fields = issue.get("fields", {})
# Basic info
key = issue.get("key", "Unknown")
summary = fields.get("summary", "No summary")
status = fields.get("status", {}).get("name", "Unknown")
issue_type = fields.get("issuetype", {}).get("name", "Unknown")
priority = fields.get("priority", {}).get("name", "Unknown")
# People
assignee = fields.get("assignee")
assignee_name = assignee.get("displayName", "Unassigned") if assignee else "Unassigned"
reporter = fields.get("reporter")
reporter_name = reporter.get("displayName", "Unknown") if reporter else "Unknown"
# Dates
created = fields.get("created", "Unknown")[:10] if fields.get("created") else "Unknown"
updated = fields.get("updated", "Unknown")[:10] if fields.get("updated") else "Unknown"
# Description (truncated)
description = fields.get("description", "No description")
if isinstance(description, dict):
# Handle Atlassian Document Format
description = "Rich text description (use get_issue for full details)"
elif len(str(description)) > 200:
description = str(description)[:200] + "..."
return f"""
š« {key}: {summary}
š Type: {issue_type} | Status: {status} | Priority: {priority}
š¤ Assignee: {assignee_name} | Reporter: {reporter_name}
š
Created: {created} | Updated: {updated}
š Description: {description}
š URL: {JIRA_BASE_URL}/browse/{key}
"""
def format_project(project: Dict[str, Any]) -> str:
"""Format a Jira project for display."""
key = project.get("key", "Unknown")
name = project.get("name", "Unknown")
project_type = project.get("projectTypeKey", "Unknown")
lead = project.get("lead", {})
lead_name = lead.get("displayName", "Unknown") if lead else "Unknown"
return f"""
š {key}: {name}
š·ļø Type: {project_type}
š¤ Lead: {lead_name}
š URL: {JIRA_BASE_URL}/browse/{key}
"""
# =============================================================================
# MCP TOOLS - Issue Management
# =============================================================================
@mcp.tool()
async def search_issues(
jql: str = "",
project: str = "",
assignee: str = "",
status: str = "",
issue_type: str = "",
priority: str = "",
max_results: int = 20
) -> str:
"""Search for Jira issues using JQL or simple filters.
Args:
jql: JQL (Jira Query Language) string for advanced searches
project: Project key or name to filter by
assignee: Assignee username, email, or 'currentUser()' for yourself
status: Status name (e.g., 'To Do', 'In Progress', 'Done')
issue_type: Issue type (e.g., 'Story', 'Bug', 'Task')
priority: Priority level (e.g., 'High', 'Medium', 'Low')
max_results: Maximum number of results to return (default: 20)
"""
try:
# Build JQL query if not provided
if not jql:
conditions = []
if project:
conditions.append(f'project = "{project}"')
if assignee:
if assignee.lower() in ['me', 'myself', 'current']:
conditions.append('assignee = currentUser()')
else:
conditions.append(f'assignee = "{assignee}"')
if status:
conditions.append(f'status = "{status}"')
if issue_type:
conditions.append(f'issuetype = "{issue_type}"')
if priority:
conditions.append(f'priority = "{priority}"')
# Ensure query is bounded (required by some Jira instances)
if not conditions:
conditions.append('created >= -30d') # Default to last 30 days
jql = " AND ".join(conditions)
# Add ordering if not present
if "order by" not in jql.lower():
jql += " ORDER BY updated DESC"
params = {
"jql": jql,
"maxResults": min(max_results, 50), # Cap at 50 for performance
"fields": "summary,status,assignee,reporter,issuetype,priority,created,updated,description"
}
result = await jira_client.make_request("GET", "search/jql", params=params)
issues = result.get("issues", [])
total = result.get("total", 0)
if not issues:
return f"No issues found matching the criteria.\nJQL used: {jql}"
formatted_issues = [format_issue(issue) for issue in issues]
header = f"Found {len(issues)} of {total} issues matching your search:\nJQL: {jql}\n"
return header + "\n" + "="*50 + "\n".join(formatted_issues)
except Exception as e:
return f"Error searching issues: {str(e)}"
@mcp.tool()
async def get_issue(issue_key: str) -> str:
"""Get detailed information about a specific Jira issue.
Args:
issue_key: The issue key (e.g., 'PROJ-123')
"""
try:
result = await jira_client.make_request("GET", f"issue/{issue_key}")
fields = result.get("fields", {})
# Get comments
comments_data = fields.get("comment", {})
comments = comments_data.get("comments", [])
# Format detailed issue info
key = result.get("key", "Unknown")
summary = fields.get("summary", "No summary")
description = fields.get("description", "No description")
# Handle Atlassian Document Format for description
if isinstance(description, dict):
# Try to extract plain text from ADF
description = extract_text_from_adf(description)
status = fields.get("status", {}).get("name", "Unknown")
issue_type = fields.get("issuetype", {}).get("name", "Unknown")
priority = fields.get("priority", {}).get("name", "Unknown")
assignee = fields.get("assignee")
assignee_name = assignee.get("displayName", "Unassigned") if assignee else "Unassigned"
reporter = fields.get("reporter")
reporter_name = reporter.get("displayName", "Unknown") if reporter else "Unknown"
created = fields.get("created", "Unknown")[:19] if fields.get("created") else "Unknown"
updated = fields.get("updated", "Unknown")[:19] if fields.get("updated") else "Unknown"
# Format comments
comments_text = ""
if comments:
comments_text = "\n\nš¬ Recent Comments:\n"
for comment in comments[-3:]: # Show last 3 comments
author = comment.get("author", {}).get("displayName", "Unknown")
created_date = comment.get("created", "")[:19] if comment.get("created") else ""
body = comment.get("body", "")
# Handle ADF format in comments
if isinstance(body, dict):
body = extract_text_from_adf(body)
comments_text += f"\n{author} ({created_date}):\n{body}\n" + "-"*30
return f"""
š« {key}: {summary}
š Details:
Type: {issue_type}
Status: {status}
Priority: {priority}
š„ People:
Assignee: {assignee_name}
Reporter: {reporter_name}
š
Dates:
Created: {created}
Updated: {updated}
š Description:
{description}
š URL: {JIRA_BASE_URL}/browse/{key}
{comments_text}
"""
except Exception as e:
return f"Error getting issue {issue_key}: {str(e)}"
def extract_text_from_adf(adf_content: Dict[str, Any]) -> str:
"""Extract plain text from Atlassian Document Format."""
if not isinstance(adf_content, dict):
return str(adf_content)
def extract_text(node):
if isinstance(node, dict):
if node.get("type") == "text":
return node.get("text", "")
elif "content" in node:
return "".join(extract_text(child) for child in node["content"])
elif "text" in node:
return node["text"]
elif isinstance(node, list):
return "".join(extract_text(item) for item in node)
return ""
return extract_text(adf_content) or "Rich text content (view in Jira for full formatting)"
@mcp.tool()
async def get_my_issues(status: str = "", max_results: int = 10) -> str:
"""Get issues assigned to the current user.
Args:
status: Filter by status (e.g., 'To Do', 'In Progress', 'Done')
max_results: Maximum number of results to return (default: 10)
"""
try:
jql = "assignee = currentUser()"
if status:
jql += f' AND status = "{status}"'
jql += " ORDER BY updated DESC"
return await search_issues(jql=jql, max_results=max_results)
except Exception as e:
return f"Error getting your issues: {str(e)}"
@mcp.tool()
async def create_issue(
project_key: str,
summary: str,
issue_type: str = "Task",
description: str = "",
priority: str = "Medium",
assignee: str = ""
) -> str:
"""Create a new Jira issue.
Args:
project_key: The project key (e.g., 'PROJ')
summary: Brief description of the issue
issue_type: Type of issue ('Story', 'Bug', 'Task', etc.)
description: Detailed description of the issue
priority: Priority level ('Highest', 'High', 'Medium', 'Low', 'Lowest')
assignee: Username or email of assignee (leave empty for unassigned)
"""
try:
# Build issue data
issue_data = {
"fields": {
"project": {"key": project_key},
"summary": summary,
"issuetype": {"name": issue_type},
"priority": {"name": priority}
}
}
# Add description if provided
if description:
issue_data["fields"]["description"] = {
"type": "doc",
"version": 1,
"content": [
{
"type": "paragraph",
"content": [
{
"type": "text",
"text": description
}
]
}
]
}
# Add assignee if provided
if assignee:
if assignee.lower() in ['me', 'myself', 'current']:
# Get current user
user_info = await jira_client.make_request("GET", "myself")
issue_data["fields"]["assignee"] = {"accountId": user_info["accountId"]}
else:
# Try to find user by email or username
try:
users = await jira_client.make_request("GET", "user/search", params={"query": assignee})
if users:
issue_data["fields"]["assignee"] = {"accountId": users[0]["accountId"]}
except:
pass # If user not found, leave unassigned
result = await jira_client.make_request("POST", "issue", json=issue_data)
issue_key = result.get("key")
issue_url = f"{JIRA_BASE_URL}/browse/{issue_key}"
return f"""
ā
Issue created successfully!
š« Key: {issue_key}
š Type: {issue_type}
š Summary: {summary}
ā” Priority: {priority}
š¤ Assignee: {assignee or 'Unassigned'}
š View issue: {issue_url}
"""
except Exception as e:
return f"Error creating issue: {str(e)}"
@mcp.tool()
async def update_issue(
issue_key: str,
status: str = "",
assignee: str = "",
priority: str = "",
summary: str = ""
) -> str:
"""Update an existing Jira issue.
Args:
issue_key: The issue key (e.g., 'PROJ-123')
status: New status (e.g., 'In Progress', 'Done')
assignee: New assignee username/email, or 'unassigned' to remove assignee
priority: New priority ('Highest', 'High', 'Medium', 'Low', 'Lowest')
summary: New summary/title for the issue
"""
try:
update_data = {"fields": {}}
# Update status (requires transition)
if status:
# Get available transitions
transitions = await jira_client.make_request("GET", f"issue/{issue_key}/transitions")
# Find matching transition
target_transition = None
for transition in transitions.get("transitions", []):
if transition["to"]["name"].lower() == status.lower():
target_transition = transition["id"]
break
if target_transition:
await jira_client.make_request(
"POST",
f"issue/{issue_key}/transitions",
json={"transition": {"id": target_transition}}
)
else:
available_statuses = [t["to"]["name"] for t in transitions.get("transitions", [])]
return f"Status '{status}' not available. Available transitions: {', '.join(available_statuses)}"
# Update other fields
if assignee:
if assignee.lower() == 'unassigned':
update_data["fields"]["assignee"] = None
elif assignee.lower() in ['me', 'myself', 'current']:
user_info = await jira_client.make_request("GET", "myself")
update_data["fields"]["assignee"] = {"accountId": user_info["accountId"]}
else:
try:
users = await jira_client.make_request("GET", "user/search", params={"query": assignee})
if users:
update_data["fields"]["assignee"] = {"accountId": users[0]["accountId"]}
except:
return f"User '{assignee}' not found"
if priority:
update_data["fields"]["priority"] = {"name": priority}
if summary:
update_data["fields"]["summary"] = summary
# Apply field updates if any
if update_data["fields"]:
await jira_client.make_request("PUT", f"issue/{issue_key}", json=update_data)
return f"""
ā
Issue {issue_key} updated successfully!
š View issue: {JIRA_BASE_URL}/browse/{issue_key}
"""
except Exception as e:
return f"Error updating issue {issue_key}: {str(e)}"
@mcp.tool()
async def add_comment(issue_key: str, comment: str) -> str:
"""Add a comment to a Jira issue.
Args:
issue_key: The issue key (e.g., 'PROJ-123')
comment: The comment text to add
"""
try:
comment_data = {
"body": {
"type": "doc",
"version": 1,
"content": [
{
"type": "paragraph",
"content": [
{
"type": "text",
"text": comment
}
]
}
]
}
}
await jira_client.make_request("POST", f"issue/{issue_key}/comment", json=comment_data)
return f"""
ā
Comment added to {issue_key}
š¬ Comment: {comment}
š View issue: {JIRA_BASE_URL}/browse/{issue_key}
"""
except Exception as e:
return f"Error adding comment to {issue_key}: {str(e)}"
@mcp.tool()
async def list_projects() -> str:
"""List all accessible Jira projects."""
try:
result = await jira_client.make_request("GET", "project")
if not result:
return "No projects found or accessible."
formatted_projects = [format_project(project) for project in result]
return f"Found {len(result)} accessible projects:\n\n" + "\n".join(formatted_projects)
except Exception as e:
return f"Error listing projects: {str(e)}"
@mcp.tool()
async def get_project_info(project_key: str) -> str:
"""Get detailed information about a specific project.
Args:
project_key: The project key (e.g., 'PROJ')
"""
try:
result = await jira_client.make_request("GET", f"project/{project_key}")
key = result.get("key", "Unknown")
name = result.get("name", "Unknown")
description = result.get("description", "No description")
project_type = result.get("projectTypeKey", "Unknown")
lead = result.get("lead", {})
lead_name = lead.get("displayName", "Unknown") if lead else "Unknown"
# Get issue types
issue_types = result.get("issueTypes", [])
issue_type_names = [it.get("name") for it in issue_types]
return f"""
š Project: {key} - {name}
š Description: {description}
š·ļø Type: {project_type}
š¤ Lead: {lead_name}
š« Available Issue Types: {', '.join(issue_type_names)}
š URL: {JIRA_BASE_URL}/browse/{key}
"""
except Exception as e:
return f"Error getting project info for {project_key}: {str(e)}"
if __name__ == "__main__":
# Run the MCP server
mcp.run(transport="stdio")
