GitHub Copilot を使った AI Agent の構築

CMDBのARN管理テーブルから取得したAWSリソースのARN情報を使って、
リソース間の依存関係を自動で分析・収集するAI Agentを実装してください。

技術スタック:
- LangGraph、LangChain
- Amazon Bedrock (Claude Sonnet 4.5)
- AWS SDK (boto3)
- Python 3.12

機能要件:
- API Endpoint: POST /service_configurations
  - パラメータ: sid, environment (どちらも必須)
- ARN管理テーブルからsid、environmentを条件にARNを検索
- 取得したCloudFront、S3、WAF、ALB、TargetGroup、ECS、RDS、ElastiCacheのARN情報を使って、Nodes、Edges情報をLLMとAgent (LangGraph) で成形
- 収集した情報をDBに保存

依存関係の取得方法 (Few-shot Examples):

### CloudFront → S3/ALB のEdge判定方法
1. CloudFront APIでドメイン名を取得
   aws cloudfront get-distribution --id {distribution-id}
2. ビヘイビア情報からOriginを取得してEdge(紐づき)を判定
   - DomainNameにs3がある場合: CloudFront → S3
   - DomainNameにALBがある場合: CloudFront → ALB

### TargetGroup → ECS のEdge判定方法
1. elbv2のAPIでターゲットのIPアドレスを取得
   aws elbv2 describe-target-health --target-group-arn {arn}
2. ECS APIでタスクのIPアドレスと照合してEdgeを作成

### ECS → RDS のEdge判定方法 (実際のアクセスではなく、セキュリティの許可で判定)
1. ECSタスクからENI (Elastic Network Interface) のIDを取得
   aws ecs describe-tasks --cluster {cluster} --tasks {task-arn}
2. ENIからECSタスクのセキュリティグループIDを取得
   aws ec2 describe-network-interfaces --network-interface-ids {eni-id}
3. RDSのセキュリティグループを取得
   aws rds describe-db-instances --db-instance-identifier {instance-id}
4. RDSセキュリティグループのインバウンドルールを取得
   aws ec2 describe-security-groups --group-ids {sg-id}
5. ECSのセキュリティグループがRDSのインバウンドルールに含まれていればEdge作成

使用予定のAWS API:
- cloudfront:get_distribution
- cloudfront:list_distributions
- wafv2:list_web_acls
- wafv2:get_web_acl
- wafv2:get_web_acl_for_resource
- elbv2:describe_target_groups
- elbv2:describe_target_health
- elbv2:describe_load_balancers
- elbv2:describe_load_balancer_attributes
- ecs:list_clusters
- ecs:describe_tasks
- ecs:describe_task_definition
- ecs:list_tasks
- ecs:describe_services
- ec2:describe_network_interfaces
- ec2:describe_security_groups
- rds:describe_db_clusters
- rds:describe_db_instances
- elasticache:describe_replication_groups
- elasticache:describe_cache_clusters

関連するDDL:
{ARN管理テーブルのDDL}
{Node情報テーブルのDDL}
{Edge情報テーブルのDDL}

重要:
- まずは現状のリポジトリ構成を理解する
- 実装を始める前に整理して、実装方針を確認してから実装を行う
- 呼び出すAWS APIや取得するNodeは増える可能性があるため、拡張性を考慮する
- コードは簡単でシンプルな内容にする

(Read files...)

リポジトリの理解ができましたので、設計方針をまとめます。

リポジトリ構造の理解
既存のプロジェクトは以下の構成になっています：

・FastAPIベースのAPIサーバー
・**Bedrock (Claude)**を使用したLLM機能
・MySQLデータベース接続機能（text_to_sql.pyで実装例あり）
・boto3でAWS APIを呼び出す機能（aws_diagram.pyで実装例あり）
・LangChainの依存関係あり（pyproject.tomlに記載）

実装方針
1. アーキテクチャ設計
ARN管理テーブルからARN情報取得 (DB)
    ↓
LangGraph Agent でAWS APIを呼び出し、Edges情報を収集
    ↓
Nodes/Edges情報をテーブルに保存 (DB)
    ↓
レスポンス返却

2. モジュール構成 (新規作成ファイル)
・`service_config_router.py`: FastAPIエンドポイント
・`service_config_agent.py`: LangGraphによるAgent実装
・`aws_tools.py`: AWS API呼び出しツール群
・`service_config_db.py`: DB操作 (ARN取得、Node/Edge保存)

3. 実装フェーズ
・Phase 1 (DB層): ARN取得とNode/Edge保存関数の実装
・Phase 2 (AWSツール層): CloudFront、ELB、ECS、RDS等のAPI呼び出し関数をLangChain Tool形式で実装
・Phase 3 (Agent層): LangGraphのStateGraphを構築し、LLMにツールを渡してEdge情報を推論
・Phase 4 (Router層): FastAPIエンドポイントとリクエスト検証・レスポンス整形

4. 設計の重要ポイント
・拡張性: ホワイトリスト方式で新しいAWS APIを追加しやすく
・シンプルさ: StateGraphで状態管理を明確に、各ツールは1関数=1API呼び出し
・データフロー: API → Agent → AWS APIs → DB保存という明確な流れ

5. 実装コード構造（擬似コード）
(コードのサンプル)

この設計方針で実装を進めてよろしいでしょうか？
確認いただけましたら、順次実装を開始します。

def create_service_config_agent() -> StateGraph:
    """システム構成収集Agentを作成"""
    workflow = StateGraph(AgentState)
    
    # ノードを追加
    workflow.add_node("initialize_nodes", initialize_nodes)
    workflow.add_node("collect_edges", collect_edges_with_llm)
    
    # エントリーポイントを設定
    workflow.set_entry_point("initialize_nodes")
    
    # 条件分岐: ノードが存在すればEdge収集、なければ終了
    workflow.add_conditional_edges(
        "initialize_nodes",
        should_collect_edges,
        {
            "collect_edges": "collect_edges",
            "end": END
        }
    )
    
    workflow.add_edge("collect_edges", END)
    
    return workflow.compile()

def collect_edges_with_llm(state: AgentState) -> AgentState:
    """LLMとツールを使用してエッジ情報を収集"""
    llm = get_llm_for_agent()
    llm_with_tools = llm.bind_tools(AWS_TOOLS)
    
    prompt = f"""
    あなたはAWSリソースの依存関係を分析するエキスパートです。

    # タスク
    以下のAWSリソース (Nodes) のARNから、リソース間の接続関係 (Edges) を推測・特定してください。
    各AWSサービスの特性と一般的なアーキテクチャパターンを理解し、適切なAWS APIを呼び出して接続を確認してください。
    
    # 利用可能なNodes
    {nodes_summary}

    # 利用可能なツール
    1. **call_aws_api**: 許可されたAWS APIを呼び出せるツール
      - リソースの詳細情報、設定、関連リソースを取得できます
    
    2. **extract_resource_id_from_arn**: ARNからリソースIDやその他の情報を抽出
      - API呼び出しに必要なパラメータ (ID、名前など) を取得できます
    
    # 使用可能なAWS API (これ以外は使用できません)
    - cloudfront:get_distribution
    - wafv2:get_web_acl_for_resource
    - elbv2:describe_target_groups
    ...

    # Edge検出の方法
    以下の観点から、リソース間の接続を推測・調査してください：
    
    ## 一般的な接続パターン
    1. **フロントエンド層**: CloudFront → S3/ALB、WAF → CloudFront/ALB、Route53ドメイン → CloudFront
    2. **ロードバランサー層**: ALB → ターゲットグループ → ECS/EC2
    3. **API層**: API Gateway → Lambda
    4. **アプリケーション層**: ECS → RDS/ElastiCache (セキュリティグループ経由) 、Lambda → RDS (セキュリティグループ経由) 
    5. **データ層**: RDS、ElastiCache

    ## 接続検出の考え方
    - **設定ベース**: リソースの設定に他のリソースのARNやIDが含まれている場合 (例: CloudFrontのOrigins設定) 
    - **ネットワークベース**: セキュリティグループのインバウンドルールで許可されている場合 (例: ECS → RDS) 
    - **サービス特性**: 各AWSサービスの役割から論理的に推測できる接続 (例: TargetGroup → ECS) 

    ## 重要な調査ポイント
    - **ARNの分析**: まずextract_resource_id_from_arnでARNを解析し、リソースタイプと必要なパラメータを特定
    - **段階的調査**: 一度に全APIを呼ばず、結果を見ながら次に必要なAPIを判断
    - **セキュリティグループ**: ECS/RDS/ElastiCacheの接続はセキュリティグループのインバウンドルールで確認
    - ECSのENI → セキュリティグループID → RDS/ElastiCacheのSGインバウンドルールに含まれるかチェック
    - **エラー対応**: 許可されていないAPIやエラーが返っても、次の調査を継続
    
    # Few-shot Examples (必ず参考にすること)
    ## Example 1: ECS → RDS の調査
    1. ECSタスクのENIを取得: call_aws_api("ecs", "describe_tasks", ...)
    2. ENIからSGを取得: call_aws_api("ec2", "describe_network_interfaces", ...)
    3. RDSのSGを取得: call_aws_api("rds", "describe_db_instances", ...)
    4. SG一致確認 → Edge作成
    ...
    
    # 重要な注意事項
    - すべてのリソースの組み合わせをチェック
    - APIエラーが出ても次の調査を継続
    - RDSの snapshot, parameter group, subnet group 等は無視
    - セキュリティグループで接続可能性を判断

    # 出力形式
    調査が完了したら、以下のJSON形式で結果を返してください：
    ```json
    {{
    "nodes": [
        {{
        "service_name": "cloudfront",
        "arn": "xxx",
        "resource": ""
        }}
    ],
    "edges": [
        {{
        "from_arn": "xxx",
        "to_arn": "xxx",
        "details": "xxx"
        }}
    ]
    }}
    ```
    """
    
    messages = [HumanMessage(content=prompt)]
    
    try:
        max_iterations = 30  # 最大反復回数
        edges = []
        
        for iteration in range(max_iterations):
            response = llm_with_tools.invoke(messages)
            messages.append(response)
            
            # ツール呼び出しがあるか確認
            if hasattr(response, 'tool_calls') and response.tool_calls:
                # ツールを実行
                tool_node = ToolNode(AWS_TOOLS)
                tool_results = tool_node.invoke({"messages": messages})
                
                # ツール結果をメッセージに追加
                messages.extend(tool_results["messages"])
                
            else:
                # ツール呼び出しがない場合、最終レスポンスとして処理
                try:
                    # 構造化レスポンス用のLLMを作成
                    structured_llm = llm.with_structured_output(GraphResult)
                    
                    # 最終結果の要約プロンプトを追加
                    final_prompt = """
                        調査が完了しました。発見したすべてのエッジと新しいノード（ドメイン名など）を
                        JSON形式で返してください。
                    """
                    messages.append(HumanMessage(content=final_prompt))
                    
                    # 構造化レスポンスを取得
                    result = structured_llm.invoke(messages)
                    
                    edges = [edge.model_dump() for edge in result.edges]
                    new_nodes = [node.model_dump() for node in result.nodes]
                    
                    # LLMが返した新しいノード（ドメイン名など）を既存のノードリストに追加
                    if new_nodes:
                        state["nodes"].extend(new_nodes)
                    
                except Exception as e:
                    # エラー時のフォールバック処理
                    ...

                break
        
        state["edges"] = edges
        state["current_step"] = "edges_collected"
        state["messages"] = messages
        
    except Exception as e:
        ...
        state["edges"] = []
    
    return state

# 許可するAWS APIのホワイトリスト
ALLOWED_AWS_APIS: Set[str] = {
    # CloudFront
    "cloudfront:get_distribution",
    "cloudfront:list_distributions",
    
    # WAF
    "wafv2:list_web_acls",
    "wafv2:get_web_acl",
    "wafv2:get_web_acl_for_resource",
    ...
}


@tool
def call_aws_api(
    service_name: str,
    method_name: str,
    parameters: Dict[str, Any],
    region: str = "ap-northeast-1"
) -> Dict[str, Any]:
    """汎用的なAWS API呼び出しツール
    
    このツールは許可されたAWS APIのみを呼び出すことができます。
    Edge情報を取得するために必要なAWS APIを呼び出してください。
    
    Args:
        service_name: AWSサービス名 (小文字) 
            許可: 'cloudfront', 'wafv2', 'elbv2', 'ecs', 'ec2', 'rds', 'elasticache', 'apigateway', 'lambda'
        method_name: 呼び出すメソッド名 (boto3のメソッド名、snake_case)
            例: 'get_distribution', 'describe_target_health', 'describe_security_groups'
        parameters: メソッドに渡すパラメータの辞書
            例: {"Id": "ABC123"} や {"GroupIds": ["sg-12345"]}
        region: AWSリージョン (デフォルト: ap-northeast-1)
    
    Returns:
        API呼び出し結果の辞書
        エラーの場合は {"error": "エラーメッセージ"} を返す
    
    許可されているAPI一覧:
        - cloudfront:get_distribution
        - cloudfront:list_distributions
        - wafv2:list_web_acls
        - wafv2:get_web_acl
        - wafv2:get_web_acl_for_resource
        - ...
    
    Examples:
        # CloudFront Distribution情報を取得
        call_aws_api(
            service_name="cloudfront",
            method_name="get_distribution",
            parameters={"Id": "ABC123"}
        )
        
        # ターゲットグループのヘルス情報を取得
        call_aws_api(
            service_name="elbv2",
            method_name="describe_target_health",
            parameters={"TargetGroupArn": "arn:aws:elasticloadbalancing:..."}
        )
        
        # セキュリティグループ情報を取得
        call_aws_api(
            service_name="ec2",
            method_name="describe_security_groups",
            parameters={"GroupIds": ["sg-12345"]}
        )
        
        # ECSタスク情報を取得
        call_aws_api(
            service_name="ecs",
            method_name="describe_tasks",
            parameters={"cluster": "my-cluster", "tasks": ["arn:aws:ecs:..."]}
        )
    """
    try:
        # ステップ1: APIホワイトリストに含まれているか検証
        is_valid, error_message = validate_aws_api(service_name, method_name)
        
        if not is_valid:
            return {
                "error": error_message,
                "error_type": "unauthorized_api",
                "allowed_apis": get_allowed_apis_list()
            }
        
        # ステップ2: クライアントを取得
        client = get_aws_client(service_name, region)
        
        # ステップ3: メソッドが存在するか確認
        if not hasattr(client, method_name):
            return {"error": error_msg, "available_methods": dir(client)}
        
        # ステップ4: メソッドを取得して実行
        method = getattr(client, method_name)
        response = method(**parameters)
        
        return response
        
    except Exception as e:
        ...