LoginSignup
2
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@BNR-Longin株式会社BinaryTech | 日本品質 × ベトナム開発、最高のバランスを。

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

Posted at

はじめに

第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ファイルに以下を確認： 
    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データを取得し、進捗やパフォーマンスを分析します。

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. サーバー起動：

    python jira_analysis_server.py

    コンソールに「Jira分析MCPサーバーを起動中: http://localhost:8134」と表示。

  2. プロジェクト分析のテスト：
    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))

    期待されるレスポンス：

    {
  "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）に以下を追加：

    {
  "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に興味がある方は、ぜひお楽しみに！

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

2
2
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?