server.py•22.4 kB
#!/usr/bin/env python3
"""
MCP Server for Jira Integration
Allows ticket creation through Jira REST API
"""
import asyncio
import os
import json
from typing import Any, Optional
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
class JiraClient:
"""Client for interacting with Jira REST API"""
def __init__(self, base_url: str, email: str, api_token: str):
self.base_url = base_url.rstrip('/')
self.email = email
self.api_token = api_token
self.auth = (email, api_token)
async def create_ticket(
self,
project_key: str,
summary: str,
description: str,
issue_type: str = "Task",
**kwargs
) -> dict[str, Any]:
"""
Create a Jira ticket
Args:
project_key: Jira project key (e.g., "PROJ")
summary: Ticket summary/title
description: Ticket description
issue_type: Type of issue (Task, Bug, Story, etc.)
**kwargs: Additional fields (priority, labels, assignee, etc.)
Returns:
Dictionary containing ticket information
"""
url = f"{self.base_url}/rest/api/3/issue"
fields = {
"project": {"key": project_key},
"summary": summary,
"description": {
"type": "doc",
"version": 1,
"content": [
{
"type": "paragraph",
"content": [
{
"type": "text",
"text": description
}
]
}
]
},
"issuetype": {"name": issue_type}
}
# Add optional fields
if "priority" in kwargs:
fields["priority"] = {"name": kwargs["priority"]}
if "labels" in kwargs:
fields["labels"] = kwargs["labels"] if isinstance(kwargs["labels"], list) else [kwargs["labels"]]
if "assignee" in kwargs:
fields["assignee"] = {"accountId": kwargs["assignee"]}
if "reporter" in kwargs:
fields["reporter"] = {"accountId": kwargs["reporter"]}
payload = {"fields": fields}
async with httpx.AsyncClient() as client:
response = await client.post(
url,
json=payload,
auth=self.auth,
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
result = response.json()
# Fetch full ticket details
ticket_key = result["key"]
ticket_info = await self.get_ticket(ticket_key)
return {
"ticket_number": ticket_key,
"ticket_url": f"{self.base_url}/browse/{ticket_key}",
"ticket_info": ticket_info
}
async def get_ticket(self, ticket_key: str) -> dict[str, Any]:
"""
Get ticket information by key
Args:
ticket_key: Jira ticket key (e.g., "PROJ-123")
Returns:
Dictionary containing ticket details
"""
url = f"{self.base_url}/rest/api/3/issue/{ticket_key}"
async with httpx.AsyncClient() as client:
response = await client.get(
url,
auth=self.auth,
headers={"Accept": "application/json"}
)
response.raise_for_status()
data = response.json()
fields = data.get("fields", {})
return {
"key": data.get("key"),
"summary": fields.get("summary"),
"description": self._extract_description(fields.get("description")),
"status": fields.get("status", {}).get("name"),
"issue_type": fields.get("issuetype", {}).get("name"),
"priority": fields.get("priority", {}).get("name") if fields.get("priority") else None,
"assignee": fields.get("assignee", {}).get("displayName") if fields.get("assignee") else None,
"reporter": fields.get("reporter", {}).get("displayName") if fields.get("reporter") else None,
"created": fields.get("created"),
"updated": fields.get("updated"),
"labels": fields.get("labels", []),
"project": fields.get("project", {}).get("key")
}
def _extract_description(self, description: Any) -> str:
"""Extract text from Jira's description format"""
if not description:
return ""
if isinstance(description, str):
return description
if isinstance(description, dict):
content = description.get("content", [])
text_parts = []
def extract_text(node):
if isinstance(node, dict):
if node.get("type") == "text":
text_parts.append(node.get("text", ""))
elif "content" in node:
for item in node["content"]:
extract_text(item)
for item in content:
extract_text(item)
return " ".join(text_parts)
return str(description)
async def create_data_quality_ticket(
self,
project_key: str,
title: str,
problem: str,
queries: Optional[str] = None,
assignee: Optional[str] = None,
priority: Optional[str] = None,
labels: Optional[list[str]] = None,
issue_type: str = "Task"
) -> dict[str, Any]:
"""
Create a standardized data quality ticket for reporting system failures
Args:
project_key: Jira project key (e.g., "PROJ")
title: Brief description of the problem
problem: Detailed description of the problem
queries: SQL queries used to diagnose the issue (optional)
assignee: Account ID of the assignee (optional)
priority: Priority level (Highest, High, Medium, Low, Lowest) (optional)
labels: List of labels to add to the ticket (optional)
Returns:
Dictionary containing ticket information
"""
# Build structured description in Jira's document format
description_content = [
{
"type": "heading",
"attrs": {"level": 2},
"content": [{"type": "text", "text": "Problem"}]
},
{
"type": "paragraph",
"content": [{"type": "text", "text": problem}]
}
]
# Add queries section if provided
if queries:
description_content.extend([
{
"type": "heading",
"attrs": {"level": 2},
"content": [{"type": "text", "text": "Queries Used to Diagnose the Issue"}]
},
{
"type": "codeBlock",
"attrs": {"language": "sql"},
"content": [
{
"type": "text",
"text": queries
}
]
}
])
fields = {
"project": {"key": project_key},
"summary": title,
"description": {
"type": "doc",
"version": 1,
"content": description_content
},
"issuetype": {"name": issue_type}
}
# Add optional fields
if assignee:
fields["assignee"] = {"accountId": assignee}
if priority:
fields["priority"] = {"name": priority}
if labels:
fields["labels"] = labels if isinstance(labels, list) else [labels]
payload = {"fields": fields}
url = f"{self.base_url}/rest/api/3/issue"
async with httpx.AsyncClient() as client:
response = await client.post(
url,
json=payload,
auth=self.auth,
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
result = response.json()
# Fetch full ticket details
ticket_key = result["key"]
ticket_info = await self.get_ticket(ticket_key)
return {
"ticket_number": ticket_key,
"ticket_url": f"{self.base_url}/browse/{ticket_key}",
"ticket_info": ticket_info
}
async def create_project(
self,
key: str,
name: str,
project_type_key: str = "software",
description: Optional[str] = None,
lead_account_id: Optional[str] = None,
project_template_key: Optional[str] = None,
assignee_type: Optional[str] = None
) -> dict[str, Any]:
"""
Create a Jira project
Args:
key: Project key (e.g., "PROJ")
name: Project name
project_type_key: Type of project (software, business, service_desk) - Default: "software"
description: Project description
lead_account_id: Account ID of the project lead (required)
project_template_key: Template key for the project (optional)
assignee_type: Default assignee type (e.g., "PROJECT_LEAD") (optional)
Returns:
Dictionary containing project information
"""
url = f"{self.base_url}/rest/api/3/project"
payload = {
"key": key,
"name": name,
"projectTypeKey": project_type_key
}
if description:
payload["description"] = description
if lead_account_id:
payload["leadAccountId"] = lead_account_id
if project_template_key:
payload["projectTemplateKey"] = project_template_key
if assignee_type:
payload["assigneeType"] = assignee_type
async with httpx.AsyncClient() as client:
response = await client.post(
url,
json=payload,
auth=self.auth,
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
result = response.json()
return {
"project_key": result.get("key"),
"project_id": result.get("id"),
"project_name": result.get("name"),
"project_url": f"{self.base_url}/browse/{result.get('key')}",
"project_type": result.get("projectTypeKey"),
"description": result.get("description"),
"lead": result.get("lead", {}).get("displayName") if result.get("lead") else None
}
async def list_projects(self) -> list[dict[str, Any]]:
"""
List all accessible Jira projects
Returns:
List of project dictionaries
"""
url = f"{self.base_url}/rest/api/3/project"
async with httpx.AsyncClient() as client:
response = await client.get(
url,
auth=self.auth,
headers={"Accept": "application/json"}
)
response.raise_for_status()
projects = response.json()
return [
{
"key": p.get("key"),
"id": p.get("id"),
"name": p.get("name"),
"project_type": p.get("projectTypeKey"),
"description": p.get("description", ""),
"lead": p.get("lead", {}).get("displayName") if p.get("lead") else None
}
for p in projects
]
# Initialize Jira client
jira_client: Optional[JiraClient] = None
def initialize_jira_client():
"""Initialize Jira client from environment variables"""
global jira_client
base_url = os.getenv("JIRA_BASE_URL")
email = os.getenv("JIRA_EMAIL")
api_token = os.getenv("JIRA_API_TOKEN")
if not all([base_url, email, api_token]):
raise ValueError(
"Missing required environment variables: "
"JIRA_BASE_URL, JIRA_EMAIL, JIRA_API_TOKEN"
)
jira_client = JiraClient(base_url, email, api_token)
# Create MCP server
server = Server("jira-mcp-server")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools"""
return [
Tool(
name="create_jira_ticket",
description="Create a new Jira ticket. Returns ticket number and full ticket information.",
inputSchema={
"type": "object",
"properties": {
"project_key": {
"type": "string",
"description": "Jira project key (e.g., 'PROJ')"
},
"summary": {
"type": "string",
"description": "Ticket summary/title"
},
"description": {
"type": "string",
"description": "Ticket description"
},
"issue_type": {
"type": "string",
"description": "Type of issue (Task, Bug, Story, Epic, etc.)",
"default": "Task"
},
"priority": {
"type": "string",
"description": "Priority level (Highest, High, Medium, Low, Lowest)",
"enum": ["Highest", "High", "Medium", "Low", "Lowest"]
},
"labels": {
"type": "array",
"items": {"type": "string"},
"description": "List of labels to add to the ticket"
},
"assignee": {
"type": "string",
"description": "Account ID of the assignee"
},
"reporter": {
"type": "string",
"description": "Account ID of the reporter"
}
},
"required": ["project_key", "summary", "description"]
}
),
Tool(
name="get_jira_ticket",
description="Get information about an existing Jira ticket by its key.",
inputSchema={
"type": "object",
"properties": {
"ticket_key": {
"type": "string",
"description": "Jira ticket key (e.g., 'PROJ-123')"
}
},
"required": ["ticket_key"]
}
),
Tool(
name="create_jira_project",
description="Create a new Jira project. Requires admin permissions.",
inputSchema={
"type": "object",
"properties": {
"key": {
"type": "string",
"description": "Project key (e.g., 'PROJ') - must be unique and uppercase"
},
"name": {
"type": "string",
"description": "Project name"
},
"project_type_key": {
"type": "string",
"description": "Type of project: 'software', 'business', or 'service_desk'",
"enum": ["software", "business", "service_desk"],
"default": "software"
},
"description": {
"type": "string",
"description": "Project description (optional)"
},
"lead_account_id": {
"type": "string",
"description": "Account ID of the project lead (required for project creation)"
},
"project_template_key": {
"type": "string",
"description": "Template key for the project (optional, e.g., 'com.atlassian.jira-core-project-templates:jira-core-simplified-task-tracking')"
},
"assignee_type": {
"type": "string",
"description": "Default assignee type (optional, e.g., 'PROJECT_LEAD')"
}
},
"required": ["key", "name", "lead_account_id"]
}
),
Tool(
name="list_jira_projects",
description="List all Jira projects accessible to the current user.",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="create_data_quality_ticket",
description="Create a standardized data quality ticket for reporting system failures. This is the standard format for reporting data issues identified through database queries (e.g., a store with 0 people).",
inputSchema={
"type": "object",
"properties": {
"project_key": {
"type": "string",
"description": "Jira project key (e.g., 'PROJ')"
},
"title": {
"type": "string",
"description": "Brief description of the problem (used as ticket title)"
},
"problem": {
"type": "string",
"description": "Detailed description of the problem that is being faced"
},
"queries": {
"type": "string",
"description": "SQL queries used to diagnose the issue (optional)"
},
"assignee": {
"type": "string",
"description": "Account ID of the assignee (optional)"
},
"priority": {
"type": "string",
"description": "Priority level (Highest, High, Medium, Low, Lowest) (optional)",
"enum": ["Highest", "High", "Medium", "Low", "Lowest"]
},
"labels": {
"type": "array",
"items": {"type": "string"},
"description": "List of labels to add to the ticket (optional)"
},
"issue_type": {
"type": "string",
"description": "Type of issue (Task, Bug, Story, etc.) - Default: 'Task'",
"default": "Task"
}
},
"required": ["project_key", "title", "problem"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Handle tool calls"""
if jira_client is None:
initialize_jira_client()
try:
if name == "create_jira_ticket":
result = await jira_client.create_ticket(**arguments)
response_text = json.dumps(result, indent=2)
return [TextContent(type="text", text=response_text)]
elif name == "get_jira_ticket":
ticket_key = arguments["ticket_key"]
ticket_info = await jira_client.get_ticket(ticket_key)
response_text = json.dumps({
"ticket_number": ticket_key,
"ticket_url": f"{jira_client.base_url}/browse/{ticket_key}",
"ticket_info": ticket_info
}, indent=2)
return [TextContent(type="text", text=response_text)]
elif name == "create_jira_project":
result = await jira_client.create_project(**arguments)
response_text = json.dumps(result, indent=2)
return [TextContent(type="text", text=response_text)]
elif name == "list_jira_projects":
projects = await jira_client.list_projects()
response_text = json.dumps({"projects": projects}, indent=2)
return [TextContent(type="text", text=response_text)]
elif name == "create_data_quality_ticket":
result = await jira_client.create_data_quality_ticket(**arguments)
response_text = json.dumps(result, indent=2)
return [TextContent(type="text", text=response_text)]
else:
raise ValueError(f"Unknown tool: {name}")
except httpx.HTTPStatusError as e:
error_msg = f"HTTP Error {e.response.status_code}: {e.response.text}"
return [TextContent(type="text", text=json.dumps({"error": error_msg}, indent=2))]
except Exception as e:
error_msg = f"Error: {str(e)}"
return [TextContent(type="text", text=json.dumps({"error": error_msg}, indent=2))]
async def main():
"""Main entry point"""
initialize_jira_client()
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())