# Directory Structure
```
├── .env.example
├── .gitignore
├── dist
│ ├── __tests__
│ │ ├── server.test.d.ts
│ │ └── server.test.js
│ ├── index.d.ts
│ ├── index.js
│ ├── schemas.d.ts
│ ├── schemas.js
│ ├── server.d.ts
│ ├── server.js
│ ├── types.d.ts
│ └── types.js
├── LICENSE
├── node_modules
│ ├── .package-lock.json
│ ├── @modelcontextprotocol
│ │ └── sdk
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ ├── bytes
│ │ ├── History.md
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── Readme.md
│ ├── content-type
│ │ ├── HISTORY.md
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ ├── depd
│ │ ├── History.md
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── Readme.md
│ ├── http-errors
│ │ ├── HISTORY.md
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ ├── iconv-lite
│ │ ├── .github
│ │ │ └── dependabot.yml
│ │ ├── Changelog.md
│ │ ├── encodings
│ │ │ ├── dbcs-codec.js
│ │ │ ├── dbcs-data.js
│ │ │ ├── index.js
│ │ │ ├── internal.js
│ │ │ ├── sbcs-codec.js
│ │ │ ├── sbcs-data-generated.js
│ │ │ ├── sbcs-data.js
│ │ │ ├── tables
│ │ │ │ ├── big5-added.json
│ │ │ │ ├── cp936.json
│ │ │ │ ├── cp949.json
│ │ │ │ ├── cp950.json
│ │ │ │ ├── eucjp.json
│ │ │ │ ├── gb18030-ranges.json
│ │ │ │ ├── gbk-added.json
│ │ │ │ └── shiftjis.json
│ │ │ ├── utf16.js
│ │ │ ├── utf32.js
│ │ │ └── utf7.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ ├── inherits
│ │ ├── inherits_browser.js
│ │ ├── inherits.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ ├── raw-body
│ │ ├── HISTORY.md
│ │ ├── index.d.ts
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ ├── README.md
│ │ └── SECURITY.md
│ ├── safer-buffer
│ │ ├── dangerous.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ ├── Porting-Buffer.md
│ │ ├── Readme.md
│ │ ├── safer.js
│ │ └── tests.js
│ ├── setprototypeof
│ │ ├── index.d.ts
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ ├── README.md
│ │ └── test
│ │ └── index.js
│ ├── statuses
│ │ ├── codes.json
│ │ ├── HISTORY.md
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ ├── toidentifier
│ │ ├── HISTORY.md
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ ├── unpipe
│ │ ├── HISTORY.md
│ │ ├── index.js
│ │ ├── LICENSE
│ │ ├── package.json
│ │ └── README.md
│ └── zod
│ ├── index.d.ts
│ ├── LICENSE
│ ├── package.json
│ └── README.md
├── package-lock.json
├── pyproject.toml
├── QUICKSTART.md
├── README.md
├── SECURITY.md
├── setup.py
├── src
│ └── mcp_jira
│ ├── __init__.py
│ ├── __main__.py
│ ├── config.py
│ ├── jira_client.py
│ ├── mcp_protocol.py
│ ├── simple_mcp_server.py
│ └── types.py
└── tests
├── __init__.py
├── conftest.py
├── test_jira_client.py
└── test_simple_mcp_server.py
```
# Files
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# Jira Configuration
JIRA_URL=https://your-domain.atlassian.net
[email protected]
JIRA_API_TOKEN=your_api_token
PROJECT_KEY=PROJ
DEFAULT_BOARD_ID=123
# Application Settings
DEBUG_MODE=false
LOG_LEVEL=INFO
# Optional: Sprint Defaults
DEFAULT_SPRINT_LENGTH=14
STORY_POINTS_FIELD=customfield_10026
MAX_SPRINT_ITEMS=50
# Performance Settings
JIRA_REQUEST_TIMEOUT=30
CACHE_TTL=300
MAX_CONCURRENT_REQUESTS=10
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
### Python ###
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Virtual Environment
venv/
ENV/
# IDE
.idea/
.vscode/
*.swp
*.swo
# Environment variables
.env
.env.local
# Logs
*.log
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# MCP Jira Integration
A simple Model Context Protocol (MCP) server for Jira that allows LLMs to act as project managers and personal assistants for teams using Jira.
## Features
### Core MCP Tools
- **create_issue** - Create new Jira issues with proper formatting
- **search_issues** - Search issues using JQL with smart formatting
- **get_sprint_status** - Get comprehensive sprint progress reports
- **get_team_workload** - Analyze team member workloads and capacity
- **generate_standup_report** - Generate daily standup reports automatically
### Project Management Capabilities
- Sprint progress tracking with visual indicators
- Team workload analysis and capacity planning
- Automated daily standup report generation
- Issue creation with proper prioritization
- Smart search and filtering of issues
## Requirements
- Python 3.8 or higher
- Jira account with API token
- MCP-compatible client (like Claude Desktop)
## Quick Setup
1. **Clone and install**:
```bash
cd mcp-jira
pip install -e .
```
2. **Configure Jira credentials** in `.env`:
```env
JIRA_URL=https://your-domain.atlassian.net
[email protected]
JIRA_API_TOKEN=your_api_token
PROJECT_KEY=PROJ
DEFAULT_BOARD_ID=123
```
3. **Run the MCP server**:
```bash
python -m mcp_jira.simple_mcp_server
```
## Usage Examples
### Creating Issues
"Create a high priority bug for the login system not working properly"
- Auto-assigns proper issue type, priority, and formatting
### Sprint Management
"What's our current sprint status?"
- Gets comprehensive progress report with metrics and visual indicators
### Team Management
"Show me the team workload for john.doe, jane.smith, mike.wilson"
- Analyzes capacity and provides workload distribution
### Daily Standups
"Generate today's standup report"
- Creates formatted report with completed, in-progress, and blocked items
## MCP Integration
### With Claude Desktop
Add to your `claude_desktop_config.json`:
```json
{
"mcpServers": {
"mcp-jira": {
"command": "python",
"args": ["-m", "mcp_jira.simple_mcp_server"],
"cwd": "/path/to/mcp-jira"
}
}
}
```
### With Other MCP Clients
The server follows the standard MCP protocol and works with any MCP-compatible client.
## Configuration
### Required Environment Variables
- `JIRA_URL` - Your Jira instance URL
- `JIRA_USERNAME` - Your Jira username/email
- `JIRA_API_TOKEN` - Your Jira API token
- `PROJECT_KEY` - Default project key for operations
### Optional Settings
- `DEFAULT_BOARD_ID` - Default board for sprint operations
- `DEBUG_MODE` - Enable debug logging (default: false)
- `LOG_LEVEL` - Logging level (default: INFO)
## Getting Jira API Token
1. Go to [Atlassian Account Settings](https://id.atlassian.com/manage-profile/security/api-tokens)
2. Click "Create API token"
3. Give it a name and copy the token
4. Use your email as username and the token as password
## Architecture
This implementation prioritizes simplicity:
- **Single MCP server file** - All tools in one place
- **Standard MCP protocol** - Uses official MCP SDK
- **Rich formatting** - Provides beautiful, readable reports
- **Error handling** - Graceful handling of Jira API issues
- **Async support** - Fast and responsive operations
## Troubleshooting
### Common Issues
1. **"No active sprint found"**
- Make sure your board has an active sprint
- Check that `DEFAULT_BOARD_ID` is set correctly
2. **Authentication errors**
- Verify your API token is correct
- Check that your username is your email address
3. **Permission errors**
- Ensure your Jira user has appropriate project permissions
- Check that the project key exists and you have access
### Debug Mode
Set `DEBUG_MODE=true` in your `.env` file for detailed logging.
## Contributing
1. Fork the repository
2. Make your changes
3. Test with your Jira instance
4. Submit a pull request
## License
MIT License - see LICENSE file
```
--------------------------------------------------------------------------------
/SECURITY.md:
--------------------------------------------------------------------------------
```markdown
# Security Policy
## Supported Versions
We release patches for security vulnerabilities. Which versions are eligible for receiving such patches depends on the CVSS v3.0 Rating:
| Version | Supported |
| ------- | ------------------ |
| 1.x.x | :white_check_mark: |
## Reporting a Vulnerability
Please report (suspected) security vulnerabilities to [email protected]. You will receive a response from us within 48 hours. If the issue is confirmed, we will release a patch as soon as possible depending on complexity but historically within a few days.
## Security Measures
1. **Authentication**: All JIRA API calls require proper authentication using API tokens
2. **Data Protection**: Sensitive data like API tokens should be provided via environment variables
3. **Input Validation**: All inputs are validated before being used in JIRA API calls
4. **Error Handling**: Errors are caught and handled appropriately without exposing sensitive information
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/mcp_jira/__init__.py:
--------------------------------------------------------------------------------
```python
"""Model Context Protocol server for Jira with Scrum Master capabilities"""
from importlib.metadata import version
__version__ = version("mcp-jira")
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mcp-jira"
version = "0.1.0"
description = "Model Context Protocol server for Jira with Scrum Master capabilities"
authors = [{name = "Warzuponus"}]
dependencies = [
"mcp>=1.0.0",
"jira>=3.5.1",
"python-dotenv>=0.19.0",
"pydantic>=2.0.0",
"pydantic-settings>=2.0.0",
"aiohttp>=3.8.0"
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.pytest.ini_options]
testpaths = ["tests"]
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Setup script for mcp-jira.
"""
from setuptools import setup, find_packages
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setup(
name="mcp-jira",
version="0.1.0",
author="Warzuponus",
description="Model Context Protocol server for Jira with project management capabilities",
long_description=long_description,
long_description_content_type="text/markdown",
url="https://github.com/warzuponus/mcp-jira",
packages=find_packages(where="src"),
package_dir={"": "src"},
classifiers=[
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
],
python_requires=">=3.8",
install_requires=[
"mcp>=1.0.0",
"jira>=3.5.1",
"python-dotenv>=0.19.0",
"pydantic>=2.0.0",
"pydantic-settings>=2.0.0",
"aiohttp>=3.8.0",
],
entry_points={
"console_scripts": [
"mcp-jira=mcp_jira.simple_mcp_server:main",
],
},
include_package_data=True,
zip_safe=False,
)
```
--------------------------------------------------------------------------------
/src/mcp_jira/__main__.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Main entry point for mcp-jira.
Allows running with `python -m mcp_jira`.
"""
import asyncio
import sys
import logging
from pathlib import Path
from .simple_mcp_server import main
from .config import get_settings, initialize_logging
def setup_logging():
"""Set up logging configuration."""
try:
settings = get_settings()
initialize_logging(settings)
except Exception as e:
# Fallback logging if config fails
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logging.getLogger(__name__).warning(f"Failed to load settings: {e}")
def check_env_file():
"""Check if .env file exists and provide helpful guidance."""
env_path = Path(".env")
if not env_path.exists():
print("⚠️ No .env file found!")
print("Please create a .env file with your Jira configuration:")
print("")
print("JIRA_URL=https://your-domain.atlassian.net")
print("[email protected]")
print("JIRA_API_TOKEN=your_api_token")
print("PROJECT_KEY=PROJ")
print("DEFAULT_BOARD_ID=123")
print("")
print("You can copy .env.example to .env and edit it with your values.")
return False
return True
if __name__ == "__main__":
print("🚀 Starting MCP Jira Server...")
setup_logging()
logger = logging.getLogger(__name__)
if not check_env_file():
sys.exit(1)
try:
logger.info("Initializing MCP Jira server...")
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.exception(f"Server failed to start: {e}")
print(f"❌ Error: {e}")
sys.exit(1)
```
--------------------------------------------------------------------------------
/QUICKSTART.md:
--------------------------------------------------------------------------------
```markdown
# Quick Start Guide
Get your MCP Jira server running in 5 minutes!
## Prerequisites
- Python 3.8+
- Jira account with API access
- Claude Desktop or another MCP client
## Step 1: Install
```bash
cd mcp-jira
pip install -e .
```
## Step 2: Configure
Create a `.env` file:
```bash
# Copy the example
cp .env.example .env
# Edit with your details
nano .env
```
Required settings:
```env
JIRA_URL=https://your-company.atlassian.net
[email protected]
JIRA_API_TOKEN=your_api_token_here
PROJECT_KEY=PROJ
```
## Step 3: Get Jira API Token
1. Go to https://id.atlassian.com/manage-profile/security/api-tokens
2. Click "Create API token"
3. Copy the token to your `.env` file
## Step 4: Test the Server
```bash
python -m mcp_jira
```
You should see: `🚀 Starting MCP Jira Server...`
## Step 5: Connect to Claude Desktop
Add to your `claude_desktop_config.json`:
```json
{
"mcpServers": {
"mcp-jira": {
"command": "python",
"args": ["-m", "mcp_jira"],
"cwd": "/path/to/mcp-jira"
}
}
}
```
## Step 6: Try It Out!
In Claude Desktop, try:
- "Create a high priority bug for login issues"
- "What's our current sprint status?"
- "Show me all issues assigned to john.doe"
- "Generate today's standup report"
## Troubleshooting
### "No .env file found"
- Make sure you created `.env` in the project root
- Copy from `.env.example` if available
### "Authentication failed"
- Check your API token is correct
- Verify your username is your email address
- Ensure the Jira URL is correct
### "No active sprint found"
- Make sure your board has an active sprint
- Set `DEFAULT_BOARD_ID` in your `.env`
### "Permission denied"
- Verify your Jira user has project access
- Check the `PROJECT_KEY` is correct
## Next Steps
- Explore all available tools with "What tools do you have?"
- Set up team workload monitoring
- Automate your daily standups
- Create custom JQL searches
Happy project managing! 🎯
```
--------------------------------------------------------------------------------
/src/mcp_jira/config.py:
--------------------------------------------------------------------------------
```python
"""
Configuration management for MCP Jira.
Handles environment variables, settings validation, and configuration defaults.
"""
from pydantic import HttpUrl, SecretStr, field_validator
from pydantic_settings import BaseSettings
from typing import Optional
import os
from functools import lru_cache
class Settings(BaseSettings):
"""
Configuration settings for the MCP Jira application.
Uses Pydantic for validation and environment variable loading.
"""
# Jira Configuration
jira_url: HttpUrl
jira_username: str
jira_api_token: SecretStr
project_key: str
default_board_id: Optional[int] = None
# Application Settings
debug_mode: bool = False
log_level: str = "INFO"
# Sprint Defaults
default_sprint_length: int = 14 # days
story_points_field: str = "customfield_10026" # Default story points field
max_sprint_items: int = 50
# Performance Settings
jira_request_timeout: int = 30 # seconds
cache_ttl: int = 300 # seconds
max_concurrent_requests: int = 10
class Config:
"""Pydantic configuration"""
env_file = ".env"
env_file_encoding = "utf-8"
case_sensitive = False
@field_validator("log_level")
@classmethod
def validate_log_level(cls, v: str) -> str:
"""Validate log level is a valid Python logging level"""
valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
upper_v = v.upper()
if upper_v not in valid_levels:
raise ValueError(f"Log level must be one of {valid_levels}")
return upper_v
@field_validator("jira_url")
@classmethod
def validate_jira_url(cls, v: HttpUrl) -> HttpUrl:
"""Ensure Jira URL is properly formatted"""
url_str = str(v)
if not url_str.endswith("/"):
url_str += "/"
return HttpUrl(url_str)
@lru_cache()
def get_settings() -> Settings:
"""
Get settings with LRU cache to avoid reading environment variables multiple times.
"""
return Settings()
def initialize_logging(settings: Settings) -> None:
"""Initialize logging configuration"""
import logging
logging.basicConfig(
level=settings.log_level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
# Set third-party loggers to WARNING to reduce noise
logging.getLogger("aiohttp").setLevel(logging.WARNING)
logging.getLogger("urllib3").setLevel(logging.WARNING)
# Example .env file template
ENV_TEMPLATE = """
# Jira Configuration
JIRA_URL=https://your-domain.atlassian.net
[email protected]
JIRA_API_TOKEN=your_api_token
PROJECT_KEY=PROJ
DEFAULT_BOARD_ID=123
# Application Settings
DEBUG_MODE=false
LOG_LEVEL=INFO
"""
def generate_env_template() -> str:
"""Generate a template .env file"""
return ENV_TEMPLATE.strip()
```
--------------------------------------------------------------------------------
/tests/test_jira_client.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the Jira client implementation.
"""
import pytest
from unittest.mock import Mock, AsyncMock, patch, MagicMock
from mcp_jira.jira_client import JiraClient
from mcp_jira.types import IssueType, Priority
@pytest.mark.asyncio
@patch('aiohttp.ClientSession')
async def test_create_issue(mock_session_class, mock_jira_client):
"""Test creating a Jira issue"""
# Set up mock session
mock_session = MagicMock()
mock_session_class.return_value.__aenter__.return_value = mock_session
mock_session_class.return_value.__aexit__.return_value = None
# Mock the POST response
mock_response = MagicMock()
mock_response.status = 201
mock_response.json = AsyncMock(return_value={"key": "TEST-1"})
mock_session.post.return_value.__aenter__.return_value = mock_response
mock_session.post.return_value.__aexit__.return_value = None
result = await mock_jira_client.create_issue(
summary="Test Issue",
description="Test Description",
issue_type=IssueType.STORY,
priority=Priority.HIGH,
story_points=5
)
assert result == "TEST-1"
@pytest.mark.asyncio
async def test_get_sprint(mock_jira_client, sample_sprint):
"""Test getting sprint details"""
sprint = await mock_jira_client.get_sprint(1)
assert sprint.id == 1
assert sprint.name == "Test Sprint"
@pytest.mark.asyncio
async def test_get_sprint_issues(mock_jira_client, sample_issue):
"""Test getting sprint issues"""
issues = await mock_jira_client.get_sprint_issues(1)
assert len(issues) > 0
assert issues[0].key == sample_issue.key
assert issues[0].summary == sample_issue.summary
@pytest.mark.asyncio
async def test_get_backlog_issues(mock_jira_client):
"""Test getting backlog issues"""
issues = await mock_jira_client.get_backlog_issues()
assert len(issues) > 0
assert all(isinstance(issue.key, str) for issue in issues)
@pytest.mark.asyncio
async def test_search_issues(mock_jira_client):
"""Test searching issues"""
jql = 'project = "TEST"'
issues = await mock_jira_client.search_issues(jql)
assert len(issues) > 0
assert all(hasattr(issue, 'key') for issue in issues)
@pytest.mark.asyncio
async def test_get_issue_history(mock_jira_client):
"""Test getting issue history"""
history = await mock_jira_client.get_issue_history("TEST-1")
assert isinstance(history, list)
@pytest.mark.asyncio
async def test_get_assigned_issues(mock_jira_client):
"""Test getting assigned issues"""
issues = await mock_jira_client.get_assigned_issues("test_user")
assert len(issues) > 0
assert all(hasattr(issue, 'assignee') for issue in issues)
@pytest.mark.asyncio
async def test_error_handling(mock_jira_client, mock_response):
"""Test error handling"""
# Mock error response
mock_jira_client._session.post = mock_response(500, {"error": "Test error"})
with pytest.raises(Exception):
await mock_jira_client.create_issue(
summary="Test Issue",
description="Test Description",
issue_type=IssueType.STORY,
priority=Priority.HIGH
)
```
--------------------------------------------------------------------------------
/src/mcp_jira/types.py:
--------------------------------------------------------------------------------
```python
"""
Type definitions and enums for the MCP Jira server.
Includes all custom types used across the application.
"""
from enum import Enum, auto
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
from datetime import datetime
class IssueType(str, Enum):
"""Jira issue types"""
STORY = "Story"
BUG = "Bug"
TASK = "Task"
EPIC = "Epic"
SUBTASK = "Sub-task"
INCIDENT = "Incident"
SERVICE_REQUEST = "Service Request"
class Priority(str, Enum):
"""Jira priority levels"""
HIGHEST = "Highest"
HIGH = "High"
MEDIUM = "Medium"
LOW = "Low"
LOWEST = "Lowest"
class SprintStatus(str, Enum):
"""Sprint statuses"""
PLANNING = "Planning"
ACTIVE = "Active"
COMPLETED = "Completed"
CANCELLED = "Cancelled"
class IssueStatus(str, Enum):
"""Issue statuses"""
TODO = "To Do"
IN_PROGRESS = "In Progress"
REVIEW = "Review"
BLOCKED = "Blocked"
DONE = "Done"
class RiskLevel(str, Enum):
"""Risk levels for sprint analysis"""
HIGH = "High"
MEDIUM = "Medium"
LOW = "Low"
class RiskType(str, Enum):
"""Types of risks that can be identified"""
SCOPE_CREEP = "Scope Creep"
RESOURCE_CONSTRAINT = "Resource Constraint"
TECHNICAL_DEBT = "Technical Debt"
DEPENDENCY_RISK = "Dependency Risk"
VELOCITY_RISK = "Velocity Risk"
CAPACITY_RISK = "Capacity Risk"
# Pydantic models for structured data
class TeamMember(BaseModel):
"""Team member information"""
username: str
display_name: str
email: Optional[str]
role: Optional[str]
capacity: Optional[float] = Field(
default=1.0,
description="Capacity as percentage (1.0 = 100%)"
)
class Issue(BaseModel):
"""Jira issue details"""
key: str
summary: str
description: Optional[str]
issue_type: IssueType
priority: Priority
status: IssueStatus
assignee: Optional[TeamMember]
story_points: Optional[float]
labels: List[str] = []
components: List[str] = []
created_at: datetime
updated_at: datetime
blocked_by: List[str] = []
blocks: List[str] = []
class Sprint(BaseModel):
"""Sprint information"""
id: int
name: str
goal: Optional[str]
status: SprintStatus
start_date: Optional[datetime]
end_date: Optional[datetime]
completed_points: float = 0
total_points: float = 0
team_members: List[TeamMember] = []
class Risk(BaseModel):
"""Risk assessment details"""
type: RiskType
level: RiskLevel
description: str
impact: str
mitigation: Optional[str]
affected_issues: List[str] = []
class SprintMetrics(BaseModel):
"""Sprint performance metrics"""
velocity: float
completion_rate: float
average_cycle_time: float
blocked_issues_count: int
scope_changes: int
team_capacity: float
burndown_ideal: Dict[str, float]
burndown_actual: Dict[str, float]
class WorkloadBalance(BaseModel):
"""Workload distribution information"""
team_member: TeamMember
assigned_points: float
issue_count: int
current_capacity: float
recommendations: List[str]
class DailyStandupItem(BaseModel):
"""Individual standup update"""
issue_key: str
summary: str
status: IssueStatus
assignee: str
blocked_reason: Optional[str]
notes: Optional[str]
time_spent: Optional[float]
# Custom exceptions
class JiraError(Exception):
"""Base exception for Jira-related errors"""
pass
class SprintError(Exception):
"""Base exception for Sprint-related errors"""
pass
class ConfigError(Exception):
"""Base exception for configuration errors"""
pass
# Type aliases for complex types
SprintPlanningResult = Dict[str, List[Issue]]
WorkloadDistribution = Dict[str, WorkloadBalance]
RiskAssessment = List[Risk]
TeamCapacityMap = Dict[str, float]
```
--------------------------------------------------------------------------------
/tests/test_simple_mcp_server.py:
--------------------------------------------------------------------------------
```python
"""
Tests for the simple MCP server implementation.
"""
import pytest
import asyncio
from unittest.mock import Mock, AsyncMock, patch
from mcp_jira.simple_mcp_server import (
list_tools, call_tool, handle_create_issue,
handle_search_issues, handle_sprint_status
)
from mcp_jira.types import IssueType, Priority, Issue, Sprint, IssueStatus, SprintStatus
from mcp.types import Tool, TextContent
@pytest.mark.asyncio
async def test_list_tools():
"""Test that tools are properly listed"""
tools = await list_tools()
assert len(tools) == 5
tool_names = [tool.name for tool in tools]
expected_tools = [
"create_issue", "search_issues", "get_sprint_status",
"get_team_workload", "generate_standup_report"
]
for expected_tool in expected_tools:
assert expected_tool in tool_names
@pytest.mark.asyncio
async def test_create_issue_tool():
"""Test create_issue tool"""
with patch('mcp_jira.simple_mcp_server.jira_client') as mock_client:
mock_client.create_issue = AsyncMock(return_value="TEST-123")
args = {
"summary": "Test Issue",
"description": "Test Description",
"issue_type": "Story",
"priority": "High"
}
result = await handle_create_issue(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "TEST-123" in result[0].text
assert "✅" in result[0].text
@pytest.mark.asyncio
async def test_search_issues_tool():
"""Test search_issues tool"""
with patch('mcp_jira.simple_mcp_server.jira_client') as mock_client:
# Mock issue data
mock_issue = Mock()
mock_issue.key = "TEST-1"
mock_issue.summary = "Test Issue"
mock_issue.status.value = "In Progress"
mock_issue.priority.value = "High"
mock_issue.assignee = None
mock_issue.story_points = 5
mock_client.search_issues = AsyncMock(return_value=[mock_issue])
args = {"jql": "project = TEST"}
result = await handle_search_issues(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "TEST-1" in result[0].text
assert "Test Issue" in result[0].text
@pytest.mark.asyncio
async def test_search_issues_no_results():
"""Test search_issues with no results"""
with patch('mcp_jira.simple_mcp_server.jira_client') as mock_client:
mock_client.search_issues = AsyncMock(return_value=[])
args = {"jql": "project = EMPTY"}
result = await handle_search_issues(args)
assert len(result) == 1
assert "No issues found" in result[0].text
@pytest.mark.asyncio
async def test_sprint_status_tool():
"""Test get_sprint_status tool"""
with patch('mcp_jira.simple_mcp_server.jira_client') as mock_client:
# Mock sprint data
mock_sprint = Mock()
mock_sprint.id = 1
mock_sprint.name = "Test Sprint"
mock_sprint.status.value = "Active"
mock_sprint.goal = "Complete features"
mock_sprint.start_date = None
mock_sprint.end_date = None
# Mock issues
mock_issue = Mock()
mock_issue.story_points = 5
mock_issue.status.value = "Done"
mock_client.get_active_sprint = AsyncMock(return_value=mock_sprint)
mock_client.get_sprint_issues = AsyncMock(return_value=[mock_issue])
args = {}
result = await handle_sprint_status(args)
assert len(result) == 1
assert isinstance(result[0], TextContent)
assert "Test Sprint" in result[0].text
assert "📊" in result[0].text
@pytest.mark.asyncio
async def test_call_tool_unknown():
"""Test calling an unknown tool"""
with patch('mcp_jira.simple_mcp_server.jira_client', Mock()):
result = await call_tool("unknown_tool", {})
assert len(result) == 1
assert "Unknown tool" in result[0].text
@pytest.mark.asyncio
async def test_call_tool_no_client():
"""Test calling tool when client is not initialized"""
with patch('mcp_jira.simple_mcp_server.jira_client', None):
result = await call_tool("create_issue", {})
assert len(result) == 1
assert "Jira client not initialized" in result[0].text
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
"""
PyTest configuration and fixtures for MCP Jira tests.
"""
import pytest
from typing import Dict, Any
import aiohttp
from datetime import datetime
from unittest.mock import MagicMock, AsyncMock
from mcp_jira.config import Settings
from mcp_jira.jira_client import JiraClient
from mcp_jira.types import Issue, Sprint, TeamMember, IssueType, Priority, IssueStatus
@pytest.fixture
def test_settings():
"""Provide test settings"""
# Mock environment variables for testing
import os
os.environ["JIRA_URL"] = "https://test-jira.example.com"
os.environ["JIRA_USERNAME"] = "test_user"
os.environ["JIRA_API_TOKEN"] = "test_token"
os.environ["PROJECT_KEY"] = "TEST"
os.environ["DEFAULT_BOARD_ID"] = "1"
return Settings()
@pytest.fixture
def mock_response():
"""Create a mock aiohttp response"""
class MockResponse:
def __init__(self, status: int, data: Dict[str, Any]):
self.status = status
self._data = data
async def json(self):
return self._data
async def text(self):
return str(self._data)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
pass
return MockResponse
@pytest.fixture
def mock_jira_client(test_settings):
"""Create a mock Jira client"""
client = JiraClient(test_settings)
# Mock the entire session to prevent HTTP calls
client._session = MagicMock()
# Mock all HTTP methods to return successful responses
async def mock_get(*args, **kwargs):
# Mock sprint response
if "sprint" in str(args[0]):
return MagicMock(status=200, json=AsyncMock(return_value={
"id": 1,
"name": "Test Sprint",
"goal": "Test Goal",
"state": "active",
"startDate": "2024-01-08T00:00:00.000Z",
"endDate": "2024-01-22T00:00:00.000Z"
}))
# Mock issue response
elif "issue" in str(args[0]):
return MagicMock(status=200, json=AsyncMock(return_value={
"issues": [{
"key": "TEST-1",
"fields": {
"summary": "Test Issue",
"description": "Test Description",
"issuetype": {"name": "Story"},
"priority": {"name": "High"},
"status": {"name": "To Do"},
"assignee": {
"name": "test_user",
"displayName": "Test User",
"emailAddress": "[email protected]"
},
"created": "2024-01-08T10:00:00.000Z",
"updated": "2024-01-08T10:00:00.000Z",
"customfield_10026": 5
}
}]
}))
async def mock_post(*args, **kwargs):
# Mock issue creation
if "issue" in str(args[0]):
return MagicMock(status=201, json=AsyncMock(return_value={"key": "TEST-1"}))
# Mock search
else:
return MagicMock(status=200, json=AsyncMock(return_value={
"issues": [{
"key": "TEST-1",
"fields": {
"summary": "Test Issue",
"description": "Test Description",
"issuetype": {"name": "Story"},
"priority": {"name": "High"},
"status": {"name": "To Do"},
"assignee": {
"name": "test_user",
"displayName": "Test User",
"emailAddress": "[email protected]"
},
"created": "2024-01-08T10:00:00.000Z",
"updated": "2024-01-08T10:00:00.000Z",
"customfield_10026": 5
}
}]
}))
client._session.get = AsyncMock(side_effect=mock_get)
client._session.post = AsyncMock(side_effect=mock_post)
return client
@pytest.fixture
def sample_issue():
"""Provide a sample issue"""
return Issue(
key="TEST-1",
summary="Test Issue",
description="Test Description",
issue_type=IssueType.STORY,
priority=Priority.HIGH,
status=IssueStatus.TODO,
assignee=TeamMember(
username="test_user",
display_name="Test User",
email="[email protected]",
role="Developer"
),
story_points=5,
labels=[],
components=[],
created_at=datetime.fromisoformat("2024-01-08T10:00:00.000"),
updated_at=datetime.fromisoformat("2024-01-08T10:00:00.000"),
blocked_by=[],
blocks=[]
)
@pytest.fixture
def sample_sprint():
"""Provide a sample sprint"""
return {
"id": 1,
"name": "Test Sprint",
"goal": "Test Goal",
"state": "active",
"startDate": "2024-01-08T00:00:00.000Z",
"endDate": "2024-01-22T00:00:00.000Z"
}
```
--------------------------------------------------------------------------------
/src/mcp_jira/mcp_protocol.py:
--------------------------------------------------------------------------------
```python
"""
MCP (Model Context Protocol) implementation for Jira integration.
Handles function definitions, resource management, and protocol handlers.
"""
from enum import Enum
from typing import Dict, Any, List, Optional, Union, Callable
from pydantic import BaseModel, Field
import asyncio
import logging
from datetime import datetime, timezone
from .types import (
Issue, Sprint, TeamMember, SprintStatus,
IssueType, Priority, Risk
)
from .jira_client import JiraClient
logger = logging.getLogger(__name__)
class MCPResourceType(str, Enum):
"""MCP Resource Types"""
ISSUE = "issue"
SPRINT = "sprint"
TEAM = "team"
METRICS = "metrics"
REPORT = "report"
class MCPFunction(BaseModel):
"""MCP Function Definition"""
name: str
description: str
resource_type: MCPResourceType
parameters: Dict[str, Any]
returns: Dict[str, Any]
handler: Optional[str] = None
class MCPContext(BaseModel):
"""MCP Context Information"""
conversation_id: str
user_id: str
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
metadata: Dict[str, Any] = Field(default_factory=dict)
class MCPRequest(BaseModel):
"""MCP Request Structure"""
function: str
parameters: Dict[str, Any]
context: MCPContext
resource_type: MCPResourceType
class MCPResponse(BaseModel):
"""MCP Response Structure"""
status: str
data: Optional[Dict[str, Any]] = None
error: Optional[str] = None
context: MCPContext
class MCPProtocolHandler:
"""
Main handler for MCP protocol implementation.
Manages resources, functions, and request processing.
"""
def __init__(self, jira_client: JiraClient):
self.jira = jira_client
self.functions: Dict[str, MCPFunction] = {}
self._register_core_functions()
def _register_core_functions(self):
"""Register core MCP functions"""
self.register_function(
MCPFunction(
name="create_issue",
description="Create a new Jira issue",
resource_type=MCPResourceType.ISSUE,
parameters={
"summary": {"type": "string", "required": True},
"description": {"type": "string", "required": True},
"issue_type": {"type": "string", "enum": [t.value for t in IssueType]},
"priority": {"type": "string", "enum": [p.value for p in Priority]},
"story_points": {"type": "number", "required": False},
"assignee": {"type": "string", "required": False}
},
returns={
"issue_key": {"type": "string"}
},
handler="handle_create_issue"
)
)
def register_function(self, function: MCPFunction):
"""Register a new MCP function"""
self.functions[function.name] = function
logger.info(f"Registered MCP function: {function.name}")
async def process_request(self, request: MCPRequest) -> MCPResponse:
"""Process an MCP request"""
try:
if request.function not in self.functions:
return MCPResponse(
status="error",
error=f"Unknown function: {request.function}",
context=request.context
)
function = self.functions[request.function]
if function.resource_type != request.resource_type:
return MCPResponse(
status="error",
error=f"Invalid resource type for function {request.function}",
context=request.context
)
handler = getattr(self, function.handler)
if not handler:
return MCPResponse(
status="error",
error=f"Handler not implemented: {function.handler}",
context=request.context
)
result = await handler(request.parameters, request.context)
return MCPResponse(
status="success",
data=result,
context=request.context
)
except Exception as e:
logger.exception(f"Error processing MCP request: {str(e)}")
return MCPResponse(
status="error",
error=str(e),
context=request.context
)
# Handler implementations
async def handle_create_issue(
self,
parameters: Dict[str, Any],
context: MCPContext
) -> Dict[str, Any]:
"""Handle create_issue function"""
issue_key = await self.jira.create_issue(
summary=parameters["summary"],
description=parameters["description"],
issue_type=IssueType(parameters["issue_type"]),
priority=Priority(parameters["priority"]),
story_points=parameters.get("story_points"),
assignee=parameters.get("assignee")
)
return {"issue_key": issue_key}
# Resource handlers
async def get_resource(
self,
resource_type: MCPResourceType,
resource_id: str
) -> Dict[str, Any]:
"""Get a resource by type and ID"""
handlers = {
MCPResourceType.ISSUE: self.jira.get_issue,
MCPResourceType.SPRINT: self.jira.get_sprint,
# Add more resource handlers...
}
handler = handlers.get(resource_type)
if not handler:
raise ValueError(f"Unknown resource type: {resource_type}")
return await handler(resource_id)
async def update_resource(
self,
resource_type: MCPResourceType,
resource_id: str,
updates: Dict[str, Any]
) -> Dict[str, Any]:
"""Update a resource"""
# Implement resource update logic
pass
```
--------------------------------------------------------------------------------
/src/mcp_jira/jira_client.py:
--------------------------------------------------------------------------------
```python
"""
JiraClient class implementation for MCP Jira.
Handles all direct interactions with the Jira API.
"""
from typing import List, Optional, Dict, Any
import aiohttp
import logging
from datetime import datetime
from base64 import b64encode
from .types import (
Issue, Sprint, TeamMember, IssueType,
Priority, IssueStatus, SprintStatus,
JiraError
)
from .config import Settings
logger = logging.getLogger(__name__)
class JiraClient:
def __init__(self, settings: Settings):
self.base_url = str(settings.jira_url).rstrip('/')
self.auth_header = self._create_auth_header(
settings.jira_username,
settings.jira_api_token
)
self.project_key = settings.project_key
self.board_id = settings.default_board_id
async def create_issue(
self,
summary: str,
description: str,
issue_type: IssueType,
priority: Priority,
story_points: Optional[float] = None,
assignee: Optional[str] = None,
labels: Optional[List[str]] = None,
components: Optional[List[str]] = None
) -> str:
"""Create a new Jira issue."""
data = {
"fields": {
"project": {"key": self.project_key},
"summary": summary,
"description": description,
"issuetype": {"name": issue_type.value},
"priority": {"name": priority.value}
}
}
if story_points:
data["fields"]["customfield_10026"] = story_points # Adjust field ID as needed
if assignee:
data["fields"]["assignee"] = {"name": assignee}
if labels:
data["fields"]["labels"] = labels
if components:
data["fields"]["components"] = [{"name": c} for c in components]
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/rest/api/2/issue",
headers=self._get_headers(),
json=data
) as response:
if response.status == 201:
result = await response.json()
return result["key"]
else:
error_data = await response.text()
raise JiraError(f"Failed to create issue: {error_data}")
async def get_sprint(self, sprint_id: int) -> Sprint:
"""Get sprint details by ID."""
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/rest/agile/1.0/sprint/{sprint_id}",
headers=self._get_headers()
) as response:
if response.status == 200:
data = await response.json()
return self._convert_to_sprint(data)
else:
error_data = await response.text()
raise JiraError(f"Failed to get sprint: {error_data}")
async def get_active_sprint(self) -> Optional[Sprint]:
"""Get the currently active sprint."""
sprints = await self._get_board_sprints(
self.board_id,
state=SprintStatus.ACTIVE
)
return sprints[0] if sprints else None
async def get_sprint_issues(self, sprint_id: int) -> List[Issue]:
"""Get all issues in a sprint."""
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/rest/agile/1.0/sprint/{sprint_id}/issue",
headers=self._get_headers()
) as response:
if response.status == 200:
data = await response.json()
return [self._convert_to_issue(i) for i in data["issues"]]
else:
error_data = await response.text()
raise JiraError(f"Failed to get sprint issues: {error_data}")
async def get_backlog_issues(self) -> List[Issue]:
"""Get all backlog issues."""
jql = f"project = {self.project_key} AND sprint is EMPTY ORDER BY Rank ASC"
return await self.search_issues(jql)
async def get_assigned_issues(self, username: str) -> List[Issue]:
"""Get issues assigned to a specific user."""
jql = f"assignee = {username} AND resolution = Unresolved"
return await self.search_issues(jql)
async def search_issues(self, jql: str) -> List[Issue]:
"""Search issues using JQL."""
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/rest/api/2/search",
headers=self._get_headers(),
json={
"jql": jql,
"maxResults": 100
}
) as response:
if response.status == 200:
data = await response.json()
return [self._convert_to_issue(i) for i in data["issues"]]
else:
error_data = await response.text()
raise JiraError(f"Failed to search issues: {error_data}")
async def get_issue_history(self, issue_key: str) -> List[Dict[str, Any]]:
"""Get the change history of an issue."""
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/rest/api/2/issue/{issue_key}/changelog",
headers=self._get_headers()
) as response:
if response.status == 200:
data = await response.json()
return self._process_changelog(data["values"])
else:
error_data = await response.text()
raise JiraError(f"Failed to get issue history: {error_data}")
# Helper methods
def _get_headers(self) -> Dict[str, str]:
"""Get headers for Jira API requests."""
return {
"Authorization": f"Basic {self.auth_header}",
"Content-Type": "application/json",
"Accept": "application/json"
}
def _create_auth_header(self, username: str, api_token: str) -> str:
"""Create base64 encoded auth header."""
auth_string = f"{username}:{api_token}"
return b64encode(auth_string.encode()).decode()
def _convert_to_issue(self, data: Dict[str, Any]) -> Issue:
"""Convert Jira API response to Issue object."""
fields = data["fields"]
return Issue(
key=data["key"],
summary=fields["summary"],
description=fields.get("description"),
issue_type=IssueType(fields["issuetype"]["name"]),
priority=Priority(fields["priority"]["name"]),
status=IssueStatus(fields["status"]["name"]),
assignee=self._convert_to_team_member(fields.get("assignee")) if fields.get("assignee") else None,
story_points=fields.get("customfield_10026"), # Adjust field ID as needed
labels=fields.get("labels", []),
components=[c["name"] for c in fields.get("components", [])],
created_at=datetime.fromisoformat(fields["created"].rstrip('Z')),
updated_at=datetime.fromisoformat(fields["updated"].rstrip('Z')),
blocked_by=[], # Would need to implement logic to determine blockers
blocks=[]
)
def _convert_to_sprint(self, data: Dict[str, Any]) -> Sprint:
"""Convert Jira API response to Sprint object."""
return Sprint(
id=data["id"],
name=data["name"],
goal=data.get("goal"),
status=SprintStatus(data["state"]),
start_date=datetime.fromisoformat(data["startDate"].rstrip('Z')) if data.get("startDate") else None,
end_date=datetime.fromisoformat(data["endDate"].rstrip('Z')) if data.get("endDate") else None
)
def _convert_to_team_member(self, data: Dict[str, Any]) -> TeamMember:
"""Convert Jira API response to TeamMember object."""
return TeamMember(
username=data["name"],
display_name=data["displayName"],
email=data.get("emailAddress")
)
def _process_changelog(self, changelog: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Process issue changelog into a more usable format."""
history = []
for entry in changelog:
for item in entry["items"]:
if item["field"] == "status":
history.append({
"from_status": item["fromString"],
"to_status": item["toString"],
"from_date": datetime.fromisoformat(entry["created"].rstrip('Z')),
"author": entry["author"]["displayName"]
})
return history
async def _get_board_sprints(
self,
board_id: int,
state: Optional[SprintStatus] = None
) -> List[Sprint]:
"""Get all sprints for a board."""
params = {"state": state.value} if state else {}
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/rest/agile/1.0/board/{board_id}/sprint",
headers=self._get_headers(),
params=params
) as response:
if response.status == 200:
data = await response.json()
return [self._convert_to_sprint(s) for s in data["values"]]
else:
error_data = await response.text()
raise JiraError(f"Failed to get board sprints: {error_data}")
```
--------------------------------------------------------------------------------
/src/mcp_jira/simple_mcp_server.py:
--------------------------------------------------------------------------------
```python
"""
Simple MCP server for Jira integration.
Implements core project management functions following MCP specification.
"""
import asyncio
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool, TextContent, ImageContent, EmbeddedResource
)
from .jira_client import JiraClient
from .config import get_settings
from .types import IssueType, Priority
logger = logging.getLogger(__name__)
# Initialize server
server = Server("mcp-jira")
# Global client (will be initialized in main)
jira_client: Optional[JiraClient] = None
@server.list_tools()
async def list_tools() -> List[Tool]:
"""List available MCP tools for Jira operations."""
return [
Tool(
name="create_issue",
description="Create a new Jira issue",
inputSchema={
"type": "object",
"properties": {
"summary": {
"type": "string",
"description": "Brief summary of the issue"
},
"description": {
"type": "string",
"description": "Detailed description of the issue"
},
"issue_type": {
"type": "string",
"enum": ["Story", "Bug", "Task", "Epic"],
"description": "Type of issue to create"
},
"priority": {
"type": "string",
"enum": ["Highest", "High", "Medium", "Low", "Lowest"],
"description": "Priority level"
},
"story_points": {
"type": "number",
"description": "Story points estimate (optional)"
},
"assignee": {
"type": "string",
"description": "Username to assign the issue to (optional)"
}
},
"required": ["summary", "description", "issue_type", "priority"]
}
),
Tool(
name="search_issues",
description="Search for Jira issues using JQL",
inputSchema={
"type": "object",
"properties": {
"jql": {
"type": "string",
"description": "JQL query to search for issues"
},
"max_results": {
"type": "number",
"description": "Maximum number of results to return (default: 20)"
}
},
"required": ["jql"]
}
),
Tool(
name="get_sprint_status",
description="Get current sprint status and progress",
inputSchema={
"type": "object",
"properties": {
"sprint_id": {
"type": "number",
"description": "Sprint ID to analyze (optional, defaults to active sprint)"
}
}
}
),
Tool(
name="get_team_workload",
description="Analyze team workload and capacity",
inputSchema={
"type": "object",
"properties": {
"team_members": {
"type": "array",
"items": {"type": "string"},
"description": "List of team member usernames to analyze"
}
},
"required": ["team_members"]
}
),
Tool(
name="generate_standup_report",
description="Generate daily standup report for the active sprint",
inputSchema={
"type": "object",
"properties": {}
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Handle tool calls for Jira operations."""
if not jira_client:
return [TextContent(type="text", text="Error: Jira client not initialized")]
try:
if name == "create_issue":
return await handle_create_issue(arguments)
elif name == "search_issues":
return await handle_search_issues(arguments)
elif name == "get_sprint_status":
return await handle_sprint_status(arguments)
elif name == "get_team_workload":
return await handle_team_workload(arguments)
elif name == "generate_standup_report":
return await handle_standup_report(arguments)
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
except Exception as e:
logger.exception(f"Error executing tool {name}: {str(e)}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
# Tool handlers
async def handle_create_issue(args: Dict[str, Any]) -> List[TextContent]:
"""Handle create_issue tool call."""
issue_key = await jira_client.create_issue(
summary=args["summary"],
description=args["description"],
issue_type=IssueType(args["issue_type"]),
priority=Priority(args["priority"]),
story_points=args.get("story_points"),
assignee=args.get("assignee")
)
return [TextContent(
type="text",
text=f"✅ Created issue {issue_key}: {args['summary']}"
)]
async def handle_search_issues(args: Dict[str, Any]) -> List[TextContent]:
"""Handle search_issues tool call."""
jql = args["jql"]
max_results = args.get("max_results", 20)
issues = await jira_client.search_issues(jql)
issues = issues[:max_results] # Limit results
if not issues:
return [TextContent(type="text", text="No issues found matching the query.")]
# Format results
result_text = f"Found {len(issues)} issues:\n\n"
for issue in issues:
status_emoji = "✅" if issue.status.value == "Done" else "🔄" if issue.status.value == "In Progress" else "📋"
priority_emoji = "🔴" if issue.priority.value in ["Highest", "High"] else "🟡" if issue.priority.value == "Medium" else "🟢"
assignee_text = f" (👤 {issue.assignee.display_name})" if issue.assignee else " (Unassigned)"
points_text = f" [{issue.story_points}pts]" if issue.story_points else ""
result_text += f"{status_emoji} **{issue.key}**: {issue.summary}\n"
result_text += f" {priority_emoji} {issue.priority.value} | {issue.status.value}{assignee_text}{points_text}\n\n"
return [TextContent(type="text", text=result_text)]
async def handle_sprint_status(args: Dict[str, Any]) -> List[TextContent]:
"""Handle get_sprint_status tool call."""
sprint_id = args.get("sprint_id")
if sprint_id:
sprint = await jira_client.get_sprint(sprint_id)
else:
sprint = await jira_client.get_active_sprint()
if not sprint:
return [TextContent(type="text", text="No active sprint found.")]
issues = await jira_client.get_sprint_issues(sprint.id)
# Calculate metrics
total_points = sum(issue.story_points for issue in issues if issue.story_points)
completed_points = sum(issue.story_points for issue in issues
if issue.story_points and issue.status.value == "Done")
in_progress_count = len([i for i in issues if i.status.value == "In Progress"])
blocked_count = len([i for i in issues if i.status.value == "Blocked"])
completion_rate = (completed_points / total_points * 100) if total_points > 0 else 0
# Build report
report = f"## 📊 Sprint Status: {sprint.name}\n\n"
report += f"**Status**: {sprint.status.value}\n"
report += f"**Goal**: {sprint.goal or 'No goal set'}\n"
if sprint.start_date and sprint.end_date:
days_remaining = (sprint.end_date - datetime.now()).days
report += f"**Duration**: {sprint.start_date.strftime('%Y-%m-%d')} to {sprint.end_date.strftime('%Y-%m-%d')}\n"
report += f"**Days Remaining**: {max(0, days_remaining)}\n"
report += f"\n### 📈 Progress\n"
report += f"- **Completion**: {completion_rate:.1f}% ({completed_points}/{total_points} points)\n"
report += f"- **Total Issues**: {len(issues)}\n"
report += f"- **In Progress**: {in_progress_count}\n"
if blocked_count > 0:
report += f"- **⚠️ Blocked**: {blocked_count}\n"
return [TextContent(type="text", text=report)]
async def handle_team_workload(args: Dict[str, Any]) -> List[TextContent]:
"""Handle get_team_workload tool call."""
team_members = args["team_members"]
report = "## 👥 Team Workload Analysis\n\n"
for member in team_members:
try:
issues = await jira_client.get_assigned_issues(member)
total_points = sum(issue.story_points for issue in issues if issue.story_points)
in_progress_count = len([i for i in issues if i.status.value == "In Progress"])
workload_emoji = "🔴" if total_points > 15 else "🟡" if total_points > 10 else "🟢"
report += f"### {workload_emoji} {member}\n"
report += f"- **Total Points**: {total_points}\n"
report += f"- **Active Issues**: {in_progress_count}\n"
report += f"- **Total Issues**: {len(issues)}\n\n"
except Exception as e:
report += f"### ❌ {member}\n"
report += f"- **Error**: Could not fetch data ({str(e)})\n\n"
return [TextContent(type="text", text=report)]
async def handle_standup_report(args: Dict[str, Any]) -> List[TextContent]:
"""Handle generate_standup_report tool call."""
active_sprint = await jira_client.get_active_sprint()
if not active_sprint:
return [TextContent(type="text", text="No active sprint found for standup report.")]
issues = await jira_client.get_sprint_issues(active_sprint.id)
# Categorize issues
yesterday = datetime.now().date()
completed_yesterday = [i for i in issues if i.status.value == "Done" and i.updated_at.date() == yesterday]
in_progress = [i for i in issues if i.status.value == "In Progress"]
blocked = [i for i in issues if i.status.value == "Blocked"]
report = f"## 🌅 Daily Standup - {datetime.now().strftime('%Y-%m-%d')}\n\n"
report += f"**Sprint**: {active_sprint.name}\n\n"
if completed_yesterday:
report += "### ✅ Completed Yesterday\n"
for issue in completed_yesterday:
assignee = issue.assignee.display_name if issue.assignee else "Unassigned"
report += f"- **{issue.key}**: {issue.summary} ({assignee})\n"
report += "\n"
if in_progress:
report += "### 🔄 In Progress\n"
for issue in in_progress:
assignee = issue.assignee.display_name if issue.assignee else "Unassigned"
points = f" [{issue.story_points}pts]" if issue.story_points else ""
report += f"- **{issue.key}**: {issue.summary} ({assignee}){points}\n"
report += "\n"
if blocked:
report += "### ⚠️ Blocked Issues\n"
for issue in blocked:
assignee = issue.assignee.display_name if issue.assignee else "Unassigned"
report += f"- **{issue.key}**: {issue.summary} ({assignee})\n"
report += "\n"
# Add quick metrics
total_points = sum(i.story_points for i in issues if i.story_points)
completed_points = sum(i.story_points for i in issues if i.story_points and i.status.value == "Done")
report += "### 📊 Sprint Metrics\n"
report += f"- **Progress**: {completed_points}/{total_points} points ({(completed_points/total_points*100):.1f}%)\n"
report += f"- **Active Issues**: {len(in_progress)}\n"
if blocked:
report += f"- **Blocked Issues**: {len(blocked)} ⚠️\n"
return [TextContent(type="text", text=report)]
async def main():
"""Main entry point for the MCP server."""
global jira_client
# Initialize settings and Jira client
settings = get_settings()
jira_client = JiraClient(settings)
logger.info("Starting MCP Jira server...")
# Run the MCP server
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())
```