Go to Qiita Advent Calendar Top
LoginSignup
4
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

Day 22

AI Agent on AWS Advent Calendar 2025

@maekawa-ari(前川)inＡＲアドバンストテクノロジ株式会社（ARI）

AWS AgentCore RuntimeでBacklog連携チャットエージェントを構築してみた

Posted at

はじめに

この記事では、AWS AgentCore RuntimeとStrands Agent Frameworkを使用して、Backlog APIと連携するAIエージェントを構築する方法を解説します。

目標

「○○さんの最近の作業を教えて」と自然言語で質問すると、BacklogのAPIを自動的に呼び出して情報を取得してくれるエージェントを実装します。

この記事で学べること

  • Strands Tools: @toolデコレータを使ったカスタムツール実装
  • 外部API連携: BacklogなどのREST APIをエージェントから呼び出す方法
  • セキュリティ: AWS Secrets Managerを使った認証情報の安全な管理
  • 実践的な実装: ユーザー名あいまい検索、エラーハンドリング、複数エンドポイントの組み合わせ

前提条件

  • OS: Ubuntu 24.04（WSL2でも可）
  • AWSアカウント: AgentCore Runtimeの利用可能なリージョン
  • AWS CLI: 設定済み
  • Python: 3.10以上
  • Backlogアカウント: プロジェクトへのアクセス権限

前回構築したAgentCore Runtime環境を使用しています。

アーキテクチャ

システム全体の構成は以下の通りです：

┌─────────────────────────────────────────────────────────────┐
│                        ユーザー                              │
│              「○○さんの最近の作業を教えて」                  　│
└────────────────────────┬────────────────────────────────────┘
                         │ 自然言語クエリ
                         ▼
┌─────────────────────────────────────────────────────────────┐
│              AWS AgentCore Runtime Environment              │
│  ┌────────────────────────────────────────────────────┐     │
│  │  my_agent.py (エントリーポイント)                　  │     │
│  │  ┌────────────────────────────────────────────┐    │     │
│  │  │  Strands Agent                             │    │     │
│  │  │  • System Prompt                           │    │     │
│  │  │  • Tools: [get_user_recent_work, ...]      │    │     │
│  │  │  • Memory: STM + LTM                       │    │     │
│  │  └────────────────────────────────────────────┘    │     │
│  └────────────────────────────────────────────────────┘     │
└────────────────────────┬────────────────────────────────────┘
                         │ Tool実行
                         ▼
┌─────────────────────────────────────────────────────────────┐
│            func/backlog.py (Python Tools)                   │
│  ┌──────────────────────┐  ┌─────────────────────────┐      │
│  │ @tool                │  │ @tool                   │      │
│  │ get_user_recent_work │  │ get_issue_details       │      │
│  │                      │  │                         │      │
│  │ 1. APIキー取得   　   │  │ 1. APIキー取得      　   │      │
│  │ 2. プロジェクトID取得 │ 　│ 2. 課題情報取得          │ 　   │
│  │ 3. 課題一覧取得       │  │ 3. コメント取得          │      │
│  │ 4. ユーザーフィルタ　 │   │ 4. マークダウン整形      │      │
│  │ 5. コメント取得       │  │                         │      │
│  │ 6. マークダウン整形  　│  │                       　│      │
│  └──────────────────────┘  └─────────────────────────┘      │
└───────┬──────────────────────────┬──────────────────────────┘
        │                          │
        │ GetSecretValue           │ HTTPS GET
        ▼                          ▼
┌──────────────────┐    ┌─────────────────────────┐
│ Secrets Manager  │    │    Backlog API          │
│ backlog/api-key  │    │ https://○○.backlog.jp   │
└──────────────────┘    └─────────────────────────┘

実装手順

ステップ1: Backlog APIキーの取得

  1. Backlogにログイン
  2. 個人設定 → API → APIキーを発行
  3. 権限設定: 読み取り専用として設定（今回はコメント読み込みのみのため）

ステップ2: AWS Secrets Managerへの登録

APIキーをSecrets Managerに安全に保存します。

aws secretsmanager create-secret \
  --name backlog/api-key \
  --description "Backlog API Key for AgentCore Runtime" \
  --secret-string '{"api_key": "YOUR_BACKLOG_API_KEY"}' \
  --region ap-northeast-1

重要ポイント:

  • Secret名: backlog/api-key（コード内で参照）
  • JSON形式: {"api_key": "..."}の形式で保存
  • リージョン: AgentCore Runtimeと同じリージョン

確認コマンド:

aws secretsmanager describe-secret \
  --secret-id backlog/api-key \
  --region ap-northeast-1

出力例:

{
    "ARN": "arn:aws:secretsmanager:ap-northeast-1:123456789012:secret:backlog/api-key-AbCdEf",
    "Name": "backlog/api-key",
    "Description": "Backlog API Key for AgentCore Runtime",
    "LastChangedDate": "2025-12-22T10:00:00.000000+09:00",
    "VersionIdsToStages": {
        "xxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx": ["AWSCURRENT"]
    }
}

ステップ3: IAM権限の設定

AgentCore RuntimeのExecution RoleにSecrets Manager読み取り権限を付与します。

3-1. ポリシーファイルの作成

cat > /tmp/backlog-secret-policy.json <<'EOF'
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": ["secretsmanager:GetSecretValue"],
    "Resource": "arn:aws:secretsmanager:ap-northeast-1:YOUR_ACCOUNT_ID:secret:backlog/api-key-*"
  }]
}
EOF

置き換え:

  • YOUR_ACCOUNT_ID: 自分のAWSアカウントID（12桁の数字）

3-2. Execution Roleの確認

grep execution_role .bedrock_agentcore.yaml

出力例:

execution_role: arn:aws:iam::123456789012:role/AmazonBedrockAgentCoreSDKRuntime-ap-northeast-1-abc12345

3-3. Roleへのポリシー追加

aws iam put-role-policy \
  --role-name AmazonBedrockAgentCoreSDKRuntime-ap-northeast-1-XXXXXXXX \
  --policy-name BacklogSecretsManagerAccess \
  --policy-document file:///tmp/backlog-secret-policy.json

※今回はCLIでポリシーを追加していますが、AWS Consoleから追加することも可能です。

ステップ4: Backlog API統合ツールの実装

プロジェクトルートにfuncディレクトリを作成し、backlog.pyを実装します。

mkdir -p func
touch func/__init__.py
touch func/backlog.py

func/backlog.py の実装

"""
Backlog API統合ツール
Secrets ManagerからAPIキーを取得し、Backlog APIを呼び出す
"""
import boto3
import json
import requests
import logging
from typing import Optional
from strands.tools import tool

logger = logging.getLogger(__name__)

# Secrets Manager client
secrets_client = boto3.client('secretsmanager', region_name='ap-northeast-1')

# Backlog API設定
BACKLOG_BASE_URL = "https://your-space.backlog.jp/api/v2"
SECRET_NAME = "backlog/api-key"
PROJECT_KEY = "YOUR_PROJECT"  # プロジェクトキーを指定

def _get_api_key() -> str:
    """Secrets ManagerからBacklog APIキーを取得"""
    try:
        response = secrets_client.get_secret_value(SecretId=SECRET_NAME)
        secret = json.loads(response['SecretString'])
        return secret['api_key']
    except Exception as e:
        logger.error(f"APIキー取得エラー: {e}")
        raise

def _get_project_id(api_key: str, project_key: str) -> Optional[int]:
    """プロジェクトキーからプロジェクトIDを取得"""
    try:
        response = requests.get(
            f"{BACKLOG_BASE_URL}/projects/{project_key}",
            params={'apiKey': api_key}
        )
        response.raise_for_status()
        project = response.json()
        return project.get('id')
    except Exception as e:
        logger.error(f"プロジェクトID取得エラー: {e}")
        return None

@tool
def get_user_recent_work(user_query: str = "ユーザー名"):
    """
    指定ユーザーの最近の作業内容を取得します

    Parameters:
        user_query: ユーザー名またはユーザーID（例: ○○、user_id）
    """
    api_key = _get_api_key()

    try:
        # プロジェクトIDを取得
        project_id = _get_project_id(api_key, PROJECT_KEY)
        if not project_id:
            return f"プロジェクト '{PROJECT_KEY}' が見つかりませんでした"

        # プロジェクトの課題一覧を取得
        issues_response = requests.get(
            f"{BACKLOG_BASE_URL}/issues",
            params={
                'apiKey': api_key,
                'projectId[]': project_id,
                'sort': 'updated',
                'order': 'desc',
                'count': 100
            }
        )
        issues_response.raise_for_status()
        all_issues = issues_response.json()

        # ユーザー名/IDであいまい検索
        user_issues = []
        matched_name = None
        matched_user_id = None
        for issue in all_issues:
            assignee = issue.get('assignee')
            if assignee:
                user_id = assignee.get('userId') or ''
                user_name = assignee.get('name') or ''
                # クエリがuserIdまたはnameに含まれるかチェック
                if user_query.lower() in user_id.lower() or user_query in user_name:
                    user_issues.append(issue)
                    if not matched_name:
                        matched_name = user_name
                        matched_user_id = user_id

        if not user_issues:
            return f"ユーザー '{user_query}' の担当課題が見つかりませんでした"

        # 最新5件に制限
        user_issues = user_issues[:5]
        user_name = matched_name or user_query

        # マークダウン形式で整形
        result = f"# {user_name}さんの直近の作業内容\n\n"
        result += f"担当課題数: {len(user_issues)}件\n\n"

        for idx, issue in enumerate(user_issues, 1):
            result += f"## {idx}. [{issue['issueKey']}] {issue['summary']}\n"
            result += f"- **状態**: {issue['status']['name']}\n"
            result += f"- **更新日時**: {issue['updated']}\n"

            # コメント取得
            comments_response = requests.get(
                f"{BACKLOG_BASE_URL}/issues/{issue['issueKey']}/comments",
                params={'apiKey': api_key, 'order': 'desc', 'count': 3}
            )

            if comments_response.status_code == 200:
                comments = comments_response.json()
                # 指定ユーザーのコメントのみ抽出
                user_comments = [
                    c for c in comments
                    if c.get('createdUser', {}).get('userId') == matched_user_id
                ]

                if user_comments:
                    result += f"- **最新のコメント**: \n"
                    for comment in user_comments[:1]:
                        content = comment.get('content', '')[:150].replace('\n', ' ')
                        result += f"  > {content}...\n"

            result += "\n"

        return result

    except requests.exceptions.RequestException as e:
        logger.error(f"Backlog API呼び出しエラー: {e}")
        return f"Backlog APIの呼び出しに失敗しました: {str(e)}"
    except Exception as e:
        logger.error(f"予期しないエラー: {e}")
        return f"エラーが発生しました: {str(e)}"

@tool
def get_issue_details(issue_key: str):
    """
    指定された課題の詳細情報とコメントを取得します

    Parameters:
        issue_key: 課題キー（例: PROJ-123）
    """
    api_key = _get_api_key()

    try:
        # 課題情報取得
        issue_response = requests.get(
            f"{BACKLOG_BASE_URL}/issues/{issue_key}",
            params={'apiKey': api_key}
        )
        issue_response.raise_for_status()
        issue = issue_response.json()

        # コメント取得
        comments_response = requests.get(
            f"{BACKLOG_BASE_URL}/issues/{issue_key}/comments",
            params={'apiKey': api_key, 'order': 'desc', 'count': 5}
        )
        comments_response.raise_for_status()
        comments = comments_response.json()

        # マークダウン形式で整形
        result = f"# [{issue['issueKey']}] {issue['summary']}\n\n"
        result += f"- **担当者**: {issue.get('assignee', {}).get('name', '未割当')}\n"
        result += f"- **状態**: {issue['status']['name']}\n"
        result += f"- **作成日**: {issue['created']}\n"
        result += f"- **更新日**: {issue['updated']}\n\n"

        if issue.get('description'):
            result += f"## 説明\n{issue['description']}\n\n"

        if comments:
            result += f"## 最新のコメント（{len(comments)}件）\n\n"
            for idx, comment in enumerate(comments, 1):
                created_user = comment.get('createdUser', {}).get('name', '不明')
                created = comment.get('created', '')
                content = comment.get('content', '')

                result += f"### {idx}. {created_user} ({created})\n"
                result += f"{content}\n\n"
                result += "---\n\n"

        return result

    except requests.exceptions.RequestException as e:
        logger.error(f"Backlog API呼び出しエラー: {e}")
        return f"課題 '{issue_key}' の取得に失敗しました: {str(e)}"
    except Exception as e:
        logger.error(f"予期しないエラー: {e}")
        return f"エラーが発生しました: {str(e)}"

実装のポイント

項目 説明
@toolデコレータ Strandsフレームワークにツールとして認識させる
docstring LLMがツールの使い方を理解するための情報源
Parameters セクション 引数の説明（Args:ではなくParameters:を使用）
あいまい検索 ユーザー名の部分一致で検索可能
エラーハンドリング try-exceptで適切なエラーメッセージを返却
ヘルパー関数 _get_api_key()、_get_project_id()で共通処理を抽出

ステップ5: エージェントへのツール登録

my_agent.pyでツールをインポートしてAgentに登録します。

my_agent.py
from bedrock_agentcore import BedrockAgentCoreApp
from strands import Agent
from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig
from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager
from func.backlog import get_user_recent_work, get_issue_details  # ← インポート
import os
import logging

logging.basicConfig(level=logging.INFO)

app = BedrockAgentCoreApp()

MEMORY_ID = os.environ.get("MEMORY_ID")

SYSTEM_PROMPT = """あなたはBacklog連携AIアシスタントです。

## 利用可能なツール
- get_user_recent_work: ユーザーの最近の作業を取得
- get_issue_details: 課題の詳細情報を取得

## 応答ガイドライン
- ユーザーの質問に対して、適切なツールを使用して回答してください
- 情報は分かりやすく整理して提示してください
- データが見つからない場合は、その旨を丁寧に伝えてください
"""

@app.entrypoint
def invoke(payload: dict):
    user_message = payload.get("prompt", "Hello!")
    session_id = payload.get("sessionId", "default_session")
    user_id = payload.get("userId", "default_user")

    if MEMORY_ID:
        agentcore_memory_config = AgentCoreMemoryConfig(
            memory_id=MEMORY_ID,
            session_id=session_id,
            actor_id=user_id
        )

        session_manager = AgentCoreMemorySessionManager(
            agentcore_memory_config=agentcore_memory_config,
            region_name="ap-northeast-1"
        )

        # ツールを登録
        agent = Agent(
            system_prompt=SYSTEM_PROMPT,
            tools=[get_user_recent_work, get_issue_details],  # ← ツール登録
            session_manager=session_manager
        )
    else:
        agent = Agent(
            system_prompt=SYSTEM_PROMPT,
            tools=[get_user_recent_work, get_issue_details]  # ← ツール登録
        )

    result = agent(user_message)
    return {"result": result.message}

if __name__ == "__main__":
    app.run()

ステップ6: デプロイ

source .venv/bin/activate
agentcore deploy

デプロイの流れ:

  1. ソースコードの圧縮
  2. S3へのアップロード
  3. CodeBuildでDockerイメージビルド
  4. ECRへのpush
  5. AgentCore Runtimeへのデプロイ

所要時間: 約5-10分

ステップ7: 動作確認

基本的な実行

# ユーザー検索
agentcore invoke '{"prompt": "○○さんの最近の作業内容を教えてください"}'

# 課題詳細
agentcore invoke '{"prompt": "PROJ-123の詳細を教えてください"}'

実行例（Claude経由での対話型実行）

本環境では、Claudeを対話インターフェースとして使用しています。ユーザーからの自然言語の質問をClaudeが受け取り、適切な形式でagentcore invokeを実行し、Runtimeからのレスポンスを整形して表示します。

ユーザー: 「○○さんの最近の作業を教えてください」
  ↓
Claude: agentcore invoke実行
  ↓
AgentCore Runtime: Backlog APIツール呼び出し
  ↓
Claude: レスポンスを整形して表示

【実行結果】
○○さんの最近の作業内容をお調べしました。

担当課題数: 3件

## 1. [PROJ-123] ユーザー認証機能の実装
- **状態**: 処理中
- **更新日時**: 2025-12-19T10:30:00Z
- **最新のコメント**:
  > OAuth2.0の実装が完了しました。次はトークンリフレッシュ機能の実装に取り組みます...

## 2. [PROJ-124] データベース設計レビュー
- **状態**: 完了
- **更新日時**: 2025-12-18T15:20:00Z

## 3. [PROJ-125] API仕様書作成
- **状態**: レビュー中
- **更新日時**: 2025-12-17T09:15:00Z

このように、対話型インターフェースを介することで、CLIコマンドを意識せずに自然な会話でBacklogの情報を取得できます。

Strands @toolデコレータの詳細解説

基本構文

backlog.py
from strands.tools import tool

@tool
def my_tool(param1: str, param2: int = 10):
    """
    ツールの説明（LLMが参照する重要な情報）

    Parameters:
        param1: パラメータ1の説明
        param2: パラメータ2の説明（デフォルト値: 10）
    """
    # 実装
    return "結果"

重要なポイント

項目 説明
@toolデコレータ 必須。これがないとツールとして認識されない
docstring 必須。LLMがツールの使い方を理解するために使用
Parametersセクション 必須。引数の説明（Args:ではなくParameters:を使用）
型ヒント 推奨。LLMが引数の型を理解しやすくなる
返り値 文字列推奨。LLMが理解しやすい形式（JSON、マークダウンなど）

間違いポイント

# ❌ NG: @toolデコレータがない
def get_data(user_id: str):
    """データ取得"""
    pass

# ❌ NG: docstringにParametersセクションがない
@tool
def get_data(user_id: str):
    """データ取得"""
    pass

# ❌ NG: Args:を使用している（Strandsは認識しない）
@tool
def get_data(user_id: str):
    """
    データ取得

    Args:
        user_id: ユーザーID
    """
    pass

# ✅ OK: すべての要件を満たしている
@tool
def get_data(user_id: str):
    """
    データ取得ツール

    Parameters:
        user_id: ユーザーID
    """
    pass

トラブルシューティング

1. ツールが認識されない

症状:

WARNING: unrecognized tool specification

原因:

  • @toolデコレータの付け忘れ
  • docstringのParameters:セクションがない
  • インデントが不正

対策:

  1. @toolデコレータを確認
  2. docstringにParameters:セクションを追加
  3. Pythonの標準インデント（4スペース）を確認

2. Secrets Manager権限エラー

症状:

AccessDeniedException: User is not authorized to perform: secretsmanager:GetSecretValue

原因:

  • IAMロールに権限が付与されていない
  • SecretのARNが間違っている
  • リージョンが一致していない

対策:

# IAMポリシーを確認
aws iam get-role-policy \
  --role-name YOUR_ROLE_NAME \
  --policy-name BacklogSecretsManagerAccess

# 出力が空の場合は権限が付与されていない
# ステップ3を再実行

3. Backlog API 403エラー

症状:

403 Client Error: Forbidden

原因:

  • APIキーが無効または期限切れ
  • APIキーの権限が不足
  • ベースURLが間違っている

対策:

  1. APIキーの確認:
# Secrets Managerから取得
aws secretsmanager get-secret-value \
  --secret-id backlog/api-key \
  --region ap-northeast-1 \
  --query SecretString \
  --output text | jq -r .api_key
  1. 直接APIをテスト:
import requests

api_key = "YOUR_API_KEY"
response = requests.get(
    "https://your-space.backlog.jp/api/v2/users/myself",
    params={'apiKey': api_key}
)
print(f"Status: {response.status_code}")
print(f"Response: {response.text}")
  1. 確認項目:
    • APIキーが正しい
    • プロジェクトキーが正しい
    • ベースURL(https://your-space.backlog.jp)が正しい
    • プロジェクトへのアクセス権限がある

4. NoneType エラー

症状:

'NoneType' object has no attribute 'lower'

原因: userIdやnameがNoneの場合の処理が不足

対策:

# ❌ NG: Noneの場合にエラーになる
user_id = assignee.get('userId', '')

# ✅ OK: Noneの場合も空文字列になる
user_id = assignee.get('userId') or ''

ログの確認方法

CloudWatch Logsでリアルタイム確認

# リアルタイム表示（最新ログを追跡）
aws logs tail \
  /aws/bedrock-agentcore/runtimes/YOUR_RUNTIME_NAME-DEFAULT \
  --log-stream-name-prefix "2025/12/22/[runtime-logs]" \
  --follow

# 最近10分のログを表示
aws logs tail \
  /aws/bedrock-agentcore/runtimes/YOUR_RUNTIME_NAME-DEFAULT \
  --since 10m

Backlog関連ログのみ抽出

aws logs tail \
  /aws/bedrock-agentcore/runtimes/YOUR_RUNTIME_NAME-DEFAULT \
  --since 10m \
  --format short | grep -E "(func.backlog|Tool|ERROR)"

まとめ

この記事では、AWS AgentCore RuntimeでBacklog APIと連携するAIエージェントを実装しました。

実装したコンポーネント

コンポーネント 説明
func/backlog.py Backlog API統合ツール（2関数）
my_agent.py エージェント本体（ツール登録）
AWS Secrets Manager APIキーの安全な管理
IAM Policy Secrets Manager読み取り権限

実装したツール

ツール名 機能
get_user_recent_work ユーザー名であいまい検索し、最近の作業を取得
get_issue_details 課題キーで課題の詳細とコメントを取得

今後の課題と展望

今回の実装を通じて、Backlogからの情報取得の基盤を構築できました。今後は以下のような拡張を検討しています：

  • 複数ツール連携: Backlogからコメント取得・要約 → Slackなどへの通知といった、複数ツールを組み合わせたワークフローの実装
  • 設定の柔軟性向上: 現在はプロジェクトキーなどを定数化していますが、実行時に動的に指定できるような柔軟な設計への改善
  • 要約機能の追加: LLMを活用したコメントの自動要約機能

応用可能なAPI

この実装パターンは他のサービスにも適用できます：

  • GitHub API: プルリクエスト・Issue管理
  • Slack API: メッセージ送信・取得
  • Jira API: チケット管理
  • Microsoft Teams: チーム連携

同じパターンでfunc/github.pyやfunc/slack.pyを作成することで、それぞれのAPI連携が可能です。

参考リンク

4
0
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
4
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?