server.py•15.5 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_project(
self,
key: str,
name: str,
project_type_key: str = "software",
description: Optional[str] = None,
lead_account_id: 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
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
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 (optional)"
}
},
"required": ["key", "name"]
}
),
Tool(
name="list_jira_projects",
description="List all Jira projects accessible to the current user.",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
]
@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)]
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())