JiraでAIを強化する | 第3章：プロジェクト分析：AIによるインサイト生成

## はじめに

第2章では、**Jira**にTicketを作成し、コメントを追加するMCPサーバーを構築し、プロジェクトタスク自動化エージェントを実現しました。これにより、AIがTicketを操作できるようになり、チームの効率が向上しました。今回は、この基盤を活用して、JiraプロジェクトのTicketデータを解析する**プロジェクト分析エージェント**を構築します。

この第3章では、TicketやSprintデータを分析し、プロジェクトの進捗、チームのパフォーマンス、潜在的な問題のインサイトを生成します。たとえば、AIが「特定のSprintで遅延が発生している」ことを検出したり、「特定のメンバーが過負荷になっている」ことを指摘したりできます。コード例とステップごとのガイドで、プロジェクト分析AIの構築を体験しましょう。さあ、始めましょう！

## プロジェクト分析エージェントとは？

プロジェクト分析エージェントは、JiraプロジェクトのTicketやSprintデータを解析し、プロジェクト管理に関するインサイトを提供するAIです。MCPサーバーを介して、以下のような機能を実現できます：
- **進捗分析**：Sprintの完了率や遅延リスクを評価。
- **パフォーマンス分析**：メンバーごとのTicket完了数や作業負荷を計算。
- **問題検出**：ボトルネックや繰り返し発生する問題を特定。

### ユースケース
- プロジェクト管理：Sprintの進捗を追跡し、期限超過のリスクを警告。
- チーム最適化：メンバーの作業負荷を評価し、タスク割り当てを提案。
- プロセス改善：繰り返し発生するバグや遅延の原因を特定。

## 開発環境の準備

第2章の環境を基に、以下の準備を行います：
- **Python 3.8以降**、**mcpライブラリ**、**requestsライブラリ**、**Claude Desktop**：これまでと同じ。
- **python-dotenv**：環境変数の管理（既にインストール済み）。
- **Jiraプロジェクト**：分析用のTicketとSprintデータを含むプロジェクト。

### Jiraのセットアップ
1. **プロジェクト準備**：
   - 第2章のプロジェクト（例：`MCP-TEST`、キー：`MCP`）を使用。
   - 複数のTicketを追加（例：Bug、Task、異なるステータス：Open、In Progress、Done）。
   - 複数のSprintを作成し、Ticketを割り当て。
   - 異なるユーザーをTicketに割り当て（例：Assignee）。
2. **Jira APIトークンの確認**：
   - 第2章のトークンを使用。
   - 権限を確認：プロジェクトのTicketとSprintデータへの読み取りアクセス。
3. **環境変数**：
   第2章の`.env`ファイルに以下を確認：
   ```bash
   JIRA_URL=https://your-domain.atlassian.net
   JIRA_EMAIL=your_email@example.com
   JIRA_API_TOKEN=your_jira_api_token
   JIRA_PROJECT_KEY=MCP
   ```

## コード例：プロジェクト分析用MCPサーバー

以下のMCPサーバーは、JiraプロジェクトのTicketとSprintデータを取得し、進捗やパフォーマンスを分析します。

```python
from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests
from requests.auth import HTTPBasicAuth
from collections import Counter
from datetime import datetime
import statistics

class JiraAnalysisServer(MCPServer):
    def __init__(self, host, port, url, email, api_token, project_key):
        super().__init__(host, port)
        self.url = url
        self.email = email
        self.api_token = api_token
        self.project_key = project_key
        self.base_url = f"{self.url}/rest/api/3"
        self.auth = HTTPBasicAuth(email, api_token)
        self.headers = {"Accept": "application/json"}
        self.register_resource("analyze_project", self.analyze_project)

    def get_tickets(self, status="all"):
        try:
            url = f"{self.base_url}/search"
            jql = f"project={self.project_key}"
            if status != "all":
                jql += f" AND status={status}"
            query = {"jql": jql, "maxResults": 100}
            response = requests.get(url, headers=self.headers, auth=self.auth, params=query)
            response.raise_for_status()
            return response.json()["issues"]
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_sprints(self):
        try:
            url = f"{self.base_url}/project/{self.project_key}/sprint"
            response = requests.get(url, headers=self.headers, auth=self.auth)
            response.raise_for_status()
            return response.json()["values"]
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def analyze_project(self, params):
        try:
            tickets = self.get_tickets()
            if isinstance(tickets, dict) and "status" in tickets:
                return tickets

            sprints = self.get_sprints()
            if isinstance(sprints, dict) and "status" in sprints:
                return sprints

            # Ticket分析
            status_counts = Counter(issue["fields"]["status"]["name"] for issue in tickets)
            assignee_counts = Counter(issue["fields"]["assignee"]["displayName"] 
                                    if issue["fields"]["assignee"] else "Unassigned" 
                                    for issue in tickets)
            issue_types = Counter(issue["fields"]["issuetype"]["name"] for issue in tickets)

            # 遅延リスク分析
            overdue_tickets = []
            for issue in tickets:
                due_date = issue["fields"].get("duedate")
                if due_date and issue["fields"]["status"]["name"] not in ["Done", "Closed"]:
                    due = datetime.strptime(due_date, "%Y-%m-%d")
                    if due < datetime.now():
                        overdue_tickets.append({
                            "key": issue["key"],
                            "summary": issue["fields"]["summary"],
                            "due_date": due_date
                        })

            # Sprint分析
            active_sprint = next((s for s in sprints if s["state"] == "ACTIVE"), None)
            sprint_completion = 0
            if active_sprint:
                sprint_tickets = self.get_tickets(f"Sprint = {active_sprint['id']}")
                if not isinstance(sprint_tickets, dict):
                    total_tickets = len(sprint_tickets)
                    done_tickets = len([t for t in sprint_tickets 
                                      if t["fields"]["status"]["name"] in ["Done", "Closed"]])
                    sprint_completion = (done_tickets / total_tickets * 100) if total_tickets > 0 else 0

            return {
                "status": "success",
                "analysis": {
                    "ticket_count": len(tickets),
                    "status_distribution": [{"status": s, "count": c} for s, c in status_counts.items()],
                    "assignee_distribution": [{"assignee": a, "count": c} for a, c in assignee_counts.items()],
                    "issue_types": [{"type": t, "count": c} for t, c in issue_types.items()],
                    "overdue_tickets": overdue_tickets,
                    "active_sprint": {
                        "name": active_sprint["name"] if active_sprint else "なし",
                        "completion_rate": f"{sprint_completion:.2f}%" if active_sprint else "N/A"
                    }
                }
            }
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = JiraAnalysisServer(
        host="localhost",
        port=8134,
        url=os.getenv("JIRA_URL"),
        email=os.getenv("JIRA_EMAIL"),
        api_token=os.getenv("JIRA_API_TOKEN"),
        project_key=os.getenv("JIRA_PROJECT_KEY")
    )
    print("Jira分析MCPサーバーを起動中: http://localhost:8134")
    server.start()
```

### コードの説明
- **get_tickets**：プロジェクトのTicketを取得（ステータスフィルター付き、最大100件）。
- **get_sprints**：プロジェクトのSprintを取得。
- **analyze_project**：TicketとSprintデータを解析し、以下のインサイトを生成：
  - Ticket総数：取得したTicketの数。
  - ステータス分布：各ステータス（Open、In Progress、Doneなど）のTicket数。
  - 担当者分布：担当者ごとのTicket数（未割り当て含む）。
  - 課題タイプ：Bug、Taskなどの分布。
  - 遅延Ticket：期限超過かつ未完了のTicket。
  - アクティブSprint：現在のSprintの名前と完了率。
- **register_resource**：プロジェクト分析をリソースとして登録。

### 前提条件
- プロジェクトに複数のTicket、Sprint、異なるステータスと担当者が存在。
- 一部のTicketに期限（Due Date）が設定済み。
- Jira APIトークンにプロジェクトへの読み取り権限がある。
- `.env`ファイルに正しい`JIRA_URL`、`JIRA_EMAIL`、`JIRA_API_TOKEN`、`JIRA_PROJECT_KEY`が設定済み。

## サーバーのテスト

サーバーが正しく動作するか確認します：

1. **サーバー起動**：
   ```bash
   python jira_analysis_server.py
   ```
   コンソールに「Jira分析MCPサーバーを起動中: http://localhost:8134」と表示。

2. **プロジェクト分析のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8134"
   payload = {
       "jsonrpc": "2.0",
       "method": "analyze_project",
       "params": {},
       "id": 1
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```
   期待されるレスポンス：
   ```json
   {
     "jsonrpc": "2.0",
     "result": {
       "status": "success",
       "analysis": {
         "ticket_count": 50,
         "status_distribution": [
           {"status": "Open", "count": 20},
           {"status": "In Progress", "count": 15},
           {"status": "Done", "count": 15}
         ],
         "assignee_distribution": [
           {"assignee": "Your Name", "count": 25},
           {"assignee": "Team Member", "count": 20},
           {"assignee": "Unassigned", "count": 5}
         ],
         "issue_types": [
           {"type": "Bug", "count": 30},
           {"type": "Task", "count": 20}
         ],
         "overdue_tickets": [
           {"key": "MCP-4", "summary": "緊急バグ", "due_date": "2025-04-20"}
         ],
         "active_sprint": {
           "name": "Sprint 1",
           "completion_rate": "60.00%"
         }
       }
     },
     "id": 1
   }
   ```

## Claude Desktopとの接続

サーバーをClaude Desktopに接続します：

1. **設定ファイルの編集**：
   Claude Desktopの設定ファイル（例：`claude_desktop_config.json`）に以下を追加：
   ```json
   {
     "mcp_servers": [
       {
         "name": "JiraAnalysisServer",
         "url": "http://localhost:8134",
         "auth": "none"
       }
     ]
   }
   ```

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   プロジェクトの分析結果を教えてください。
   ```
   レスポンス例：
   ```
   プロジェクト MCP の分析結果：
   - Ticket総数：50
   - ステータス分布：Open（20）、In Progress（15）、Done（15）
   - 担当者分布：Your Name（25）、Team Member（20）、未割り当て（5）
   - 課題タイプ：Bug（30）、Task（20）
   - 遅延Ticket：MCP-4（緊急バグ、期限：2025-04-20）
   - アクティブSprint：Sprint 1（完了率：60.00%）
   ```

## 実装のコツと注意点

- **データ品質**：TicketやSprintが少ない場合、分析結果が制限される。十分なデータ（例：50件以上）を用意。
- **レートリミティング**：Jira APIの制限（例：600リクエスト/分、クラウドインスタンスによる）に注意。
- **セキュリティ**：本番環境では、`auth: none`を避け、トークン認証を導入。
- **テスト**：テスト用プロジェクトを作成し、本番データに影響を与えない。
- **拡張性**：大量のデータを処理する場合、キャッシュ（例：Redis）やNLPライブラリ（例：spaCy）でTicketサマリー分析を強化。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- 特定の課題タイプ（例：Bug）のTicketだけを分析するフィルター。
- Ticketのコメント内容を分析し、キーワード（例：「緊急」）を抽出。
- 分析結果をJiraに新しいTicketとして投稿するツール。

## まとめと次のステップ

この第3章では、JiraのTicketとSprintデータを活用してプロジェクト分析エージェントを構築しました。進捗やパフォーマンスを分析することで、AIがプロジェクト管理のインサイトを提供し、プロセス改善を支援できるようになりました。

次の第4章では、JiraのWebhookを活用して**リアルタイム管理AI**を構築します。たとえば、AIが新しいTicketやステータス変更をリアルタイムで検知し、通知や応答を生成します。リアルタイム管理AIに興味がある方は、ぜひお楽しみに！

---

役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！