トップ > Bedrock AgentCore Gateway > Kiro (IDE) から AWS ALB ログを自然言語で直接解析する！ Bedrock AgentCore Gateway × MCP

Kiro (IDE) から AWS ALB ログを自然言語で直接解析する！ Bedrock AgentCore Gateway × MCP

Bedrock AgentCore Gateway Application Load Balancer ALB
記事タイトルとURLをコピーする

構成図：

はじめに

「昨日のエラーログ見せて」とチャットに打つだけで、裏側で勝手にS3からデータを検索してきてくれる。そんな環境ができたら最高じゃないですか？

今回は、2025年の最新技術トレンドである MCP (Model Context Protocol) と、AWSの Amazon Bedrock AgentCore Gateway を組み合わせて、話題のAI IDE 「Kiro」 から直接 AWS 上の ALB アクセスログを解析できる環境を自分の環境に構築しました。

いくつものエラー（OpenAPI定義がうまく動作しない、kiro に mcp.json が複数あって間違える・・、そして ローカル mcp サーバーの作成でめちゃハマる・・）を乗り越えて完成した構成を共有します。

作ったもの

Kiro (IDE) のチャット欄で以下のように話しかけると、特定の日付のログを検索・要約してくれます。

私: 「2025年8月4日のログを確認して」
Kiro: 「ログを取得しました。ステータス200で https://... へのアクセスが確認できました...」

サンプル1 「2025年8月4日のALBログを取得して」

「2025年8月4日のALBログを取得して」への回答：

2025年8月4日のALBログを取得します。

2025年8月4日のALBログを取得しました。以下が結果です：

time | request_url | elb_status_code | client_ip
2025-08-04T07:00:13.452168Z | https://youkoso.karukozaka46.click:443/ | 200 | 106.156.155.230
2025-08-04T06:36:57.344352Z | https://13.114.164.123:443/ | 200 | 106.156.155.230
2025-08-04T06:16:06.627725Z | https://youkoso.karukozaka46.click:443/ | 200 | 106.156.155.230
2025-08-04T06:13:30.881536Z | https://youkoso.karukozaka46.click:443/ | 200 | 106.156.155.230
2025-08-04T06:13:14.140283Z | https://13.114.164.123:443/ | 200 | 106.156.155.230
2025-08-04T06:13:10.084923Z | https://13.114.164.123:443/ | 200 | 106.156.155.230
2025-08-04T06:13:08.334186Z | https://13.114.164.123:443/ | 200 | 106.156.155.230
2025-08-04T05:52:00.455460Z | https://13.114.164.123:443/ | 200 | 106.156.155.230
2025-08-04T05:49:54.832415Z | https://youkoso.karukozaka46.click:443/ | 200 | 106.156.155.230

この日のログでは、すべてのリクエストがステータスコード200（成功）で、同一のクライアントIP（106.156.155.230）から2つの異なるドメインへのアクセスが記録されています：

  • youkoso.karukozaka46.click
  • 13.114.164.123（IPアドレス直接アクセス）

アクセス時間は05:49から07:00の間に集中しています。

サンプル2 「2025年8月4日のALBログのうち、ステータスコードが 200 のもので不審なものがないか教えて」

※ ステータスコード200 (正常) の不正形式リクエストは攻撃が成功したことを意味する可能性があります。これは ALB のアクセスログ解析時によく調査する項目です。

「2025年8月4日のALBログのうち、ステータスコードが 200 のもので不審なものがないか教えて」への回答：

2025年8月4日のALBログ（ステータスコード200）を分析して、不審な点がないか確認します。

取得したログを詳しく見ると、以下の特徴があります：

正常に見える点：

  • すべてHTTPS（443ポート）でのアクセス
  • ステータスコード200で正常応答
  • アクセス時間が早朝（5:49-7:00 JST）で一般的な時間帯

注意すべき点：

  1. 単一IPからの集中アクセス

    • すべて同じクライアントIP（106.156.155.230）からのアクセス
    • 短時間（約1時間10分）で9回のアクセス

  2. 2つの異なるアクセス先

    • youkoso.karukozaka46.click - ドメイン名でのアクセス
    • 13.114.164.123 - IPアドレス直接アクセス

  3. IPアドレス直接アクセス

    • 13.114.164.123への直接アクセスは、通常のWebブラウジングでは珍しい
    • セキュリティスキャンやボット活動の可能性

推奨調査項目：

  • クライアントIP 106.156.155.230 の地理的位置と所属組織
  • 13.114.164.123 が正当なサーバーかどうか
  • 同じIPからの他の日のアクセスパターン
  • User-Agentやリファラー情報（ログに含まれていれば）

現時点では明確に悪意のある活動とは断定できませんが、IPアドレス直接アクセスと単一IPからの集中アクセスは監視対象として適切です。

私が ALB のパブリック IPアドレスへの直接アクセス（IPスキャンでの攻撃を想定）をあえて試していたので、それが怪しいログとして検出されています。

パターン3 「怪しいログがないか教えて」

「怪しいログがないか教えて」への回答：

怪しいログを調べるために、エラーステータスコードのログも確認してみます。

2025年8月4日のログを総合的に分析した結果、以下の怪しい活動が確認されました：

🚨 不審なアクセスパターン

1. 複数のIPアドレスへの探索的アクセス

  • 13.114.164.123 (200応答)
  • 54.249.83.233 (403応答) ← 新たに発見
  • youkoso.karukozaka46.click (200応答)

2. 典型的なスキャン行動

  • 同一クライアントIP（106.156.155.230）から複数のターゲットを探索
  • IPアドレス直接アクセス（通常のブラウジングでは稀）
  • favicon.icoへのアクセス（自動化ツールの典型的な挙動）

3. アクセス拒否の発生

  • 54.249.83.233への2回のアクセスが403 Forbiddenで拒否
  • このIPは何らかのセキュリティ制御で保護されている可能性

🔍 セキュリティ上の懸念

この活動パターンはネットワークスキャンやポートスキャンの可能性が高く、以下の対策を推奨します：

  • クライアントIP 106.156.155.230 のブロック検討
  • 54.249.83.233 のセキュリティ設定確認
  • 同様のスキャン活動の継続監視
  • WAF（Web Application Firewall）の導入検討

明らかに自動化されたスキャン活動と判断されます。

こういった形で、ALB のアクセスログを直接読んで怪しいリクエストを探す処理をエージェントが代行してくれます。

アーキテクチャ構成

全体のデータの流れは以下の通りです。

  1. Client (IDE): Kiro (VS CodeベースのAIエディタ)。 Kiro のクラウドサーバー（AI）と通信し自然言語処理を行う。
  2. Interface: ローカルの Python ブリッジスクリプト (MCP Server)
  3. Gateway: Amazon Bedrock AgentCore Gateway (MCP対応のエンドポイント)
  4. Compute: AWS Lambda (SQL生成・実行)
  5. Query Engine: Amazon Athena
  6. Storage: Amazon S3 (ALBログ)

全体としては、「あなたのPC（ローカル）」 から 「AWSクラウドの奥深く（S3）」 までリクエストがバケツリレーのように渡され、データを持って帰ってくる流れになります。

1. 指示の入力とAIの判断 (Kiro Client ⇄ Kiro Cloud)

  • 起点 (Local): あなたが Kiro のチャット欄に「昨日のエラーログを見せて」と入力します。
  • クラウドでの思考 (Internet): Kiro アプリ（Client）は、あなたの入力を Kiro のクラウドサーバー（AI） に送信します。AI はその意図を理解し、「ユーザーの要望を満たすには、手元の MCP Bridge Server ツール (Python ブリッジスクリプト)を引数 2025-08-04 で使う必要がある」と判断します。
  • 実行指令 (Return to Local): クラウド上の AI は直接ツールを実行せず、あなたの PC にある Kiro アプリに対して「このツールを実行してくれ」という指示書（JSON）だけを送り返します。

2. ローカルでのツール実行とAWSへの接続 (Kiro Client → Interface → Gateway)

  • ローカル呼び出し: 指示を受け取った Kiro アプリは、あなたの PC 内で動いている Python ブリッジスクリプト (Interface) を起動します。ここが重要なポイントで、実際のツール実行は常にあなたの PC から行われます。
  • 認証と送信: 命令を受け取った Pythonスクリプトが「クライアントID」と「機密鍵（Secret）」を持って、Cognito の発行所（URL）にアクセスして「通行手形（認証トークン）」を取得します。そして、インターネット経由で Amazon Bedrock AgentCore Gateway (Gateway) にリクエストを送信します。

3. リクエストの受け付けと実行 (Gateway → Compute)

  • 窓口業務: AWS 側の入り口である Gateway がリクエストを受け取ります。「これはログ解析の依頼だな」と確認し、紐付けられている AWS Lambda (Compute) を起動します。 AgentCore Gateway に Lambda をターゲット登録する際に「ツール定義（JSONスキーマ）」内の descriptionに記載した内容から、そのように判断します。
    Description: "指定された日付のALBアクセスログをAthenaを使って検索・取得します" と記載しています。
  • クエリの組み立て: Lambda は「プログラム」として動作します。受け取った日付（例: 2025-08-04）を元に、データベースを検索するための命令文（SQL）を自動生成します。

4. データの検索と抽出 (Compute → Query Engine → Storage)

  • 検索命令: Lambda は、検索エンジンの Amazon Athena (Query Engine) に「この SQL でデータを探してきて！」と命令します。
  • ファイルスキャン: Athena は、実際のデータ置き場である Amazon S3 (Storage) にある大量のログファイルを高速に読み込みます。
  • 抽出: 条件（日付やエラーコード）に一致するデータだけを抜き出し、Lambda に結果を返します。

5. 結果の解析と表示 (Interface → Kiro Client ⇄ Kiro Cloud)

  • バケツリレー (帰り): 抽出された生データ（ログの羅列）は、来た道を戻ります。 Athena → Lambda → Gateway → Pythonスクリプト → Kiro アプリ(Local)
  • AIによる要約 (Internet): Kiro アプリは、受け取ったログデータを再び Kiro クラウド に送信します。「ツールを実行したらこんな結果が出たよ」と報告するためです。
  • 最終出力: クラウドの AI がログの内容を解析し、「エラーはありませんでした」や「特定のIPからのアクセスが多いです」といった自然な日本語を作成して返信します。これを Kiro アプリが表示して完了です。

1. バックエンド構築：Athena & Lambda

まずは、ログを検索するための足回りです。

Athena のテーブル作成

テーブル作成 DDL が公式ドキュメントにあるので、参考にし、テーブルを作成しました。

Athena のコンソールでクエリできるようになりました。

Lambda の実装

Athena にクエリを発行する Lambda 関数を Python (Boto3) で実装しました。 ここで一つハマりポイントがありました。「従来の Bedrock Agent」と「新しい AgentCore Gateway (MCP)」では、Lambda に渡されるイベント構造（JSON）が違う という点です。

これを回避するために、どちらの形式でも受け取れる「二刀流」の実装にしました。

import boto3
import time
import os
import json

# クライアント初期化
athena_client = boto3.client('athena')

# 環境変数
DATABASE = os.environ.get('ATHENA_DATABASE')
TABLE = os.environ.get('ATHENA_TABLE')
OUTPUT_LOCATION = os.environ.get('ATHENA_OUTPUT_BUCKET')

def lambda_handler(event, context):
    print(f"Received event: {json.dumps(event)}")
    
    # --- 1. 入力パラメータの解析 (新旧両対応) ---
    # デフォルト値
    agent_params = {}
    is_legacy_agent = False
    
    # 分岐処理
    if 'parameters' in event and 'actionGroup' in event:
        # A. 従来の Bedrock Agent 形式
        agent_params = {p['name']: p['value'] for p in event.get('parameters', [])}
        is_legacy_agent = True
        print("Mode: Legacy Bedrock Agent")
    else:
        # B. 新しい AgentCore Gateway (MCP) 形式
        # event 自体がパラメータの辞書になっている
        agent_params = event
        is_legacy_agent = False
        print("Mode: AgentCore Gateway (MCP)")

    # --- 2. SQLクエリの構築 ---
    target_date = agent_params.get('date')
    status_code = agent_params.get('status_code')
    source_ip = agent_params.get('source_ip')
    limit = 20
    
    # 基本のSELECT文
    query = f"""
        SELECT time, request_url, elb_status_code, client_ip
        FROM "{DATABASE}"."{TABLE}"
        WHERE 1=1
    """
    
    # 条件の追加
    if target_date:
        query += f" AND date_parse(split_part(time, 'T', 1), '%Y-%m-%d') = date '{target_date}'"
    
    if status_code and status_code.isdigit():
        query += f" AND elb_status_code = {status_code}"
        
    if source_ip:
        query += f" AND client_ip LIKE '{source_ip}%'"
        
    query += f" ORDER BY time DESC LIMIT {limit}"
    
    print(f"Executing SQL: {query}")

    # --- 3. Athenaクエリ実行 ---
    response_text = ""
    try:
        execution_id = start_query(query)
        status, error_reason = wait_for_query_to_complete(execution_id)
        
        if status == 'SUCCEEDED':
            response_text = get_query_results_as_text(execution_id)
        else:
            response_text = f"Error: Query finished with status {status}. Reason: {error_reason}"

    except Exception as e:
        print(f"Exception: {str(e)}")
        response_text = f"System Error: {str(e)}"

    # --- 4. レスポンスの返却 (新旧両対応) ---
    if is_legacy_agent:
        # A. 従来の Bedrock Agent 用の複雑なラッパー
        return {
            "messageVersion": "1.0",
            "response": {
                "actionGroup": event['actionGroup'],
                "function": event['function'],
                "functionResponse": {
                    "responseBody": {
                        "TEXT": {
                            "body": response_text
                        }
                    }
                }
            }
        }
    else:
        # B. AgentCore Gateway (MCP) 用のシンプルレスポンス
        # テキストをそのまま返すか、辞書で返すとGatewayが解釈してくれます
        return {
            "result": response_text
        }

# --- Helper Functions (変更なし) ---

def start_query(query_string):
    response = athena_client.start_query_execution(
        QueryString=query_string,
        QueryExecutionContext={'Database': DATABASE},
        ResultConfiguration={'OutputLocation': OUTPUT_LOCATION}
    )
    return response['QueryExecutionId']

def wait_for_query_to_complete(execution_id):
    max_retries = 30
    for _ in range(max_retries):
        response = athena_client.get_query_execution(QueryExecutionId=execution_id)
        status_info = response['QueryExecution']['Status']
        state = status_info['State']
        if state == 'SUCCEEDED':
            return 'SUCCEEDED', None
        elif state == 'FAILED':
            return 'FAILED', status_info.get('StateChangeReason', 'Unknown')
        elif state == 'CANCELLED':
            return 'CANCELLED', 'Query was cancelled'
        time.sleep(1)
    return 'TIMEOUT', 'Query timed out'

def get_query_results_as_text(execution_id):
    response = athena_client.get_query_results(QueryExecutionId=execution_id)
    rows = response['ResultSet']['Rows']
    if len(rows) <= 1:
        return "検索条件に一致するログは見つかりませんでした。"
    formatted_rows = []
    for row in rows:
        data = [col.get('VarCharValue', 'NULL') for col in row['Data']]
        formatted_rows.append(" | ".join(data))
    return "\n".join(formatted_rows)

検証環境でLimit = 20 になっているので、もう少し増やしてもいいかもしれません。

テストデータを入れて実行すると、SQL クエリ発行し実行に成功しました。

新しい AgentCore Gateway (MCP) からのリクエストを想定したペイロードのテストデータ

{
  "date": "2025-08-04",
  "status_code": "200"
}

成功：

従来の Bedrock Agent からのリクエストを想定したペイロードのテストデータ

{
  "messageVersion": "1.0",
  "function": "function_name_placeholder",
  "actionGroup": "action_group_placeholder",
  "parameters": [
    {
      "name": "date",
      "type": "STRING",
      "value": "2025-08-04"
    },
    {
      "name": "status_code",
      "type": "STRING",
      "value": "200"
    }
  ],
  "agent": {
    "name": "my-agent",
    "version": "DRAFT",
    "id": "AGENTID12345"
  }
}

成功：

2. ゲートウェイ構築：Bedrock AgentCore Gateway

2025年の新機能、AgentCore Gateway を使用しました。 これにより、Lambda を安全な API として公開し、MCP クライアントから接続可能にします。

Bedrock の画面ではなく専用のサービス画面があります。

  • ターゲット設定: Lambda を指定し ARN を入れます。

  • ターゲットスキーマ: Lambda を使用するときは OpenAPI ではなく、公式ドキュメントに沿って JSON を書く必要がありました。

参考：AWS Lambda function targets - Amazon Bedrock AgentCore

[
  {
    "description": "指定された日付とステータスコードに基づいて、Athena経由でALBログを取得します。",
    "inputSchema": {
      "properties": {
        "date": {
          "description": "検索対象の日付 (YYYY-MM-DD形式)",
          "type": "string"
        },
        "status_code": {
          "description": "HTTPステータスコード (例: 200, 500)",
          "type": "string"
        }
      },
      "required": [
        "date"
      ],
      "type": "object"
    },
    "name": "get_alb_logs"
  }
]

ツール定義（JSONスキーマ） ：Bedrock Agent に「ALBログ検索機能」を認識させるための設定です。 特に重要なのが description です。AI はこの自然言語の説明文を読み、「ユーザーがログ解析を求めている＝このツールを起動すべき」という判断（セマンティック・ルーティング）を行います。また、inputSchema によって、会話の中から「日付」と「ステータスコード」を自動的に抽出して Lambda に渡すよう指示しています。日付を必須パラメータにしているため、何も書かないと「いつのログですか？」と AI に聞かれます。

3. クライアント構築：ローカル MCP ブリッジ (bridge.py)

Kiro などの AI IDE は標準で MCP に対応していますが、AWS Bedrock AgentCore Gateway に直接接続するには、いくつかの「壁」がありました。

  1. 認証の壁: AWS Gateway はパブリックに公開されていますが、アクセスには Cognito のアクセストークン (または IAM SigV4 署名) が必要です。Kiro の標準設定画面には、動的にトークンを取得してヘッダーに埋め込む機能はありません。
  2. プロトコルの壁: Kiro はローカルプロセスとの通信 (Stdio) を得意としますが、AWS Gateway は HTTP 上での MCP 通信 (SSE/POST) を行います。

そこで、Kiro と AWS の間を取り持つ 「中継用スクリプト (bridge.py)」 を作成しました。このスクリプトは、以下の二つの顔を持つ「プロキシ（代理人）」として機能します。

  • 対 Kiro: 標準入出力 (Stdio) で話す MCP サーバー
  • 対 AWS: 認証トークンを持って HTTP で話す API クライアント

完成したコード (bridge.py)

Python のメタプログラミング（関数の動的生成）と FastMCP を組み合わせた、そこそこ堅牢な実装です。

起動時にツールリストを MCP サーバーに登録します。

sequenceDiagram
    autonumber
    participant Bridge as 🐍 Bridge.py (起動時)
    participant Gateway as ☁️ AWS Gateway

    Note right of Bridge: 1. サーバー起動
    Bridge->>Gateway: list_tools() <br>「今使えるツール一覧をください」
    Gateway-->>Bridge: ツールリスト返却<br>[get_alb_logs, restart_ec2, ...]

    loop 取得した全ツールに対して
        Bridge->>Bridge: 関数を動的生成 (partial)<br>AWSへの転送用ラッパーを作成
        Bridge->>Bridge: Docstring (説明文) をコピー
        Bridge->>Bridge: ローカルMCPサーバーに登録
    end

    Note right of Bridge: 2. 準備完了 (待機状態)
    
    participant Kiro as 🖥️ Kiro
    Kiro->>Bridge: 接続開始
    Bridge-->>Kiro: 「私はこれらのツールを持っています」<br>(AWSからコピーしたリストを提示)

ユーザーから質問が来た際のシーケンスです。

sequenceDiagram
    participant User as 👤 User
    participant Kiro as 🖥️ Kiro (MCP Client)
    box "Local PC" #f9f9f9
        participant Bridge as 🐍 Bridge.py
    end
    box "AWS Cloud" #eceff1
        participant Cognito as 🔐 Cognito
        participant Gateway as ⛩️ AgentCore Gateway
        participant Lambda as ⚡ Lambda (Athena)
    end

    User->>Kiro: 「昨日のログを見せて」
    Kiro->>Kiro: 必要なツールを判断 (get_alb_logs)
    
    Note over Kiro, Bridge: 🔌 Stdio通信 (標準入出力)
    Kiro->>Bridge: Tool実行要求 (JSON)

    Note over Bridge: ⚙️ 内部処理: 引数のアンラップ

    Note over Bridge, Cognito: 🌐 HTTP通信 (インターネット)
    Bridge->>Cognito: 認証リクエスト (Client ID/Secret)
    Cognito-->>Bridge: Access Token 発行

    Bridge->>Gateway: POST /mcp (Token付与)
    Gateway->>Lambda: イベント発火
    Lambda->>Lambda: SQL生成 & Athena検索
    Lambda-->>Gateway: 検索結果 (JSON)
    Gateway-->>Bridge: HTTPレスポンス

    Note over Kiro, Bridge: 🔌 Stdio通信
    Bridge-->>Kiro: 結果を返す (Stdio)
    
    Kiro->>User: 自然言語で回答
import asyncio
import requests
import sys
import logging
import traceback
from functools import partial
from typing import Any

from mcp.server.fastmcp import FastMCP
from mcp.client.streamable_http import streamablehttp_client
from mcp import ClientSession

# --- 0. ログ設定: ライブラリの余計なログを消す ---
# 標準出力に余計な文字が混ざるとMCP通信が切断されるため必須
logging.basicConfig(level=logging.ERROR)

# --- 1. AWS設定 (環境変数での管理を推奨) ---
CLIENT_ID = "<YOUR_COGNITO_CLIENT_ID>"
CLIENT_SECRET = "<YOUR_COGNITO_CLIENT_SECRET>"
TOKEN_URL = "https://your-domain.auth.ap-northeast-1.amazoncognito.com/oauth2/token"
GATEWAY_URL = "https://alb-log-gateway-xxxx.gateway.bedrock-agentcore.ap-northeast-1.amazonaws.com/mcp"

# 動的であることを示すサーバー名
mcp = FastMCP("AWS-Dynamic-Bridge")

# --- ログ出力用ヘルパー ---
def log(msg):
    # 重要: Kiroとの通信は標準出力(stdout)を使うため、
    # 人間用のログは標準エラー出力(stderr)に逃がす
    print(f"[Bridge Log] {msg}", file=sys.stderr)

def fetch_access_token():
    """Cognitoからアクセストークンを取得"""
    try:
        response = requests.post(
            TOKEN_URL,
            data={
                "grant_type": "client_credentials",
                "client_id": CLIENT_ID,
                "client_secret": CLIENT_SECRET
            },
            headers={'Content-Type': 'application/x-www-form-urlencoded'}
        )
        response.raise_for_status()
        return response.json()['access_token']
    except Exception as e:
        log(f"Token Error: {e}")
        raise

async def proxy_execution(tool_name: str, **kwargs: Any) -> str:
    """
    Kiroからのリクエストを受け取り、AWSへ転送する汎用関数
    """
    log(f"1. Raw args received: {kwargs}")
    
    # --- マトリョーシカ対策: 引数の皮むき処理 ---
    actual_args = kwargs
    # Kiroが気を利かせて {"kwargs": {"date": ...}} という形で送ってくる場合があるため、
    # もし "kwargs" キーの中に辞書が入っていたら、中身を取り出してフラットにする。
    if 'kwargs' in kwargs and isinstance(kwargs['kwargs'], dict):
        actual_args = kwargs['kwargs']
        log(f"2. Unwrapped args (sending to AWS): {actual_args}")
    else:
        log("2. No unwrapping needed.")
    # --------------------------------

    try:
        token = fetch_access_token()
        headers = {"Authorization": f"Bearer {token}"}

        # AWS Gateway (MCP Server) にクライアントとして接続
        async with streamablehttp_client(url=GATEWAY_URL, headers=headers) as (read, write, _):
            async with ClientSession(read, write) as session:
                await session.initialize()
                
                # 指定されたツール名と、皮をむいた引数で実行
                result = await session.call_tool(
                    name=tool_name,
                    arguments=actual_args 
                )

                output = []
                for content in result.content:
                    if content.type == 'text':
                        output.append(content.text)
                return "\n".join(output)

    except Exception as e:
        log(f"Execution Error: {e}")
        traceback.print_exc(file=sys.stderr)
        return f"Error executing {tool_name}: {str(e)}"

async def setup_dynamic_tools():
    """
    起動時に一度だけAWSに接続し、ツール一覧をコピーして登録する
    """
    log("Connecting to AWS to fetch available tools...")
    try:
        token = fetch_access_token()
        headers = {"Authorization": f"Bearer {token}"}
        
        async with streamablehttp_client(url=GATEWAY_URL, headers=headers) as (read, write, _):
            async with ClientSession(read, write) as session:
                await session.initialize()
                
                # AWSにある全ツールを取得
                remote_tools_list = await session.list_tools()
                log(f"Found {len(remote_tools_list.tools)} tools from AWS.")
                
                for tool in remote_tools_list.tools:
                    # 長い名前 (target-xxx___get_alb_logs) を扱いやすく整形
                    clean_name = tool.name.split('___')[-1]
                    log(f" - Registering: {clean_name}")
                    
                    safe_desc = tool.description if tool.description else "No description provided."
                    
                    # partialを使って「特定のツール名専用の分身関数」を作る
                    wrapper_func = partial(proxy_execution, tool_name=tool.name)
                    
                    # FastMCPの仕様に合わせ、手動で関数名とDocstringを付与
                    wrapper_func.__name__ = clean_name
                    wrapper_func.__doc__ = safe_desc
                    
                    # 動的にサーバーに登録！
                    mcp.tool(
                        name=clean_name,
                        description=safe_desc
                    )(wrapper_func)

    except Exception as e:
        log(f"Failed to fetch tools: {e}")
        traceback.print_exc(file=sys.stderr)

if __name__ == "__main__":
    # 1. サーバー起動前にAWSへ聞きに行く
    asyncio.run(setup_dynamic_tools())
    
    # 2. 準備ができたらサーバー開始
    log("Bridge Server is ready. Waiting for Kiro...")
    mcp.run()

技術的なこだわりポイント

このコードには、実運用で発生する課題を解決するための工夫が詰め込まれています。

1. 動的ツール登録 (Dynamic Tool Registration)

特定のツール（get_alb_logs）をコードにハードコードするのではなく、起動時に Gateway に「今どんなターゲットが登録されている？」と問い合わせ、それらをループ処理で動的に登録しています。

これにより、将来 AWS 側（AgentCore Gateway）の設定で、新しい Lambda を 「ターゲット」として追加 するだけで、Python コードを一切書き換えることなく Kiro で新しい機能が使えるようになります。

sequenceDiagram
    autonumber
    participant Bridge as 🐍 Bridge.py (起動時)
    participant Gateway as ☁️ AWS Gateway

    Note right of Bridge: 1. サーバー起動
    Bridge->>Gateway: list_tools() <br>「今使えるツール一覧をください」
    Gateway-->>Bridge: ツールリスト返却<br>[get_alb_logs, restart_ec2, ...]

    loop 取得した全ツールに対して
        Bridge->>Bridge: 関数を動的生成 (partial)<br>AWSへの転送用ラッパーを作成
        Bridge->>Bridge: Docstring (説明文) をコピー
        Bridge->>Bridge: ローカルMCPサーバーに登録
    end

    Note right of Bridge: 2. 準備完了 (待機状態)
    
    participant Kiro as 🖥️ Kiro
    Kiro->>Bridge: 接続開始
    Bridge-->>Kiro: 「私はこれらのツールを持っています」<br>(AWSからコピーしたリストを提示)

2. マトリョーシカ問題の解決 (Unwrapping)

MCP クライアント（Kiro）によっては、引数を {"kwargs": {"date": "..."}} のように丁寧に包んで送ってくる場合があります。しかし、AWS 側はフラットな {"date": "..."} を期待しているため、そのまま送るとエラーになります。 そこで、「kwargs というキーがあったら中身を取り出して皮を剥く」 というアンラップ処理を実装し、互換性を確保しました。

graph TD
    %% ノードの定義
    Kiro[("🖥️ Kiro (Client)<br>引数をラップして送信")]
    Bridge[("🐍 Bridge.py<br>(皮むき処理)")]
    AWS[("☁️ AWS Lambda<br>フラットなJSONを期待")]

    %% スタイリング
    classDef error fill:#ffcccc,stroke:#ff0000,color:#000;
    classDef success fill:#ccffcc,stroke:#00aa00,color:#000;
    classDef logic fill:#e1f5fe,stroke:#01579b,color:#000;

    %% フロー
    Kiro -- "kwargsで包まれたJSON" --> Bridge
    
    subgraph Data_Flow [データ構造の変化]
        direction TB
        Input["❌ 入力データ:<br>{ 'kwargs': { 'date': '2025-08-04' } }"]
        Process{{"kwargsキーはあるか？"}}
        Action["アンラップ実行 (皮を剥く)"]
        Output["✅ 出力データ:<br>{ 'date': '2025-08-04' }"]
    end

    Bridge -- 内部処理 --> Data_Flow
    Data_Flow --> AWS

    %% エラー時の仮想フロー
    Input -.->|そのまま送信すると...| Error["💥 AWS側でバリデーションエラー"]:::error
    Output -->|これならOK| Success["⭕ 正常に処理開始"]:::success

    class Process,Action logic

3. 安全なログ出力

MCP は標準入出力（stdin/stdout）を使って通信するため、print() でデバッグログを出力すると、通信データ（JSON）にゴミが混ざって切断されてしまいます。 これを防ぐため、ログ出力関数 log() を作成し、全てのログを 標準エラー出力 (sys.stderr) に逃がすように設計しています。

コードの主要部分解説

1. 起動時のセットアップ (setup_dynamic_tools)

async def setup_dynamic_tools():
    # ...AWS接続処理...
    remote_tools_list = await session.list_tools()
    
    for tool in remote_tools_list.tools:
        # 名前を綺麗にする (target-xxx___toolname -> toolname)
        clean_name = tool.name.split('___')[-1]

        # partialで「プロキシ関数の分身」を作成
        wrapper_func = partial(proxy_execution, tool_name=tool.name)
        
        # FastMCPが関数として認識できるように名札をつける
        wrapper_func.__name__ = clean_name
        wrapper_func.__doc__ = tool.description
        
        # 登録！
        mcp.tool(name=clean_name, ...)(wrapper_func)

この部分がシステムの心臓部です。 partial を使うことで、汎用的な転送関数 proxy_execution から、「get_alb_logs 専用の転送関数」「restart_ec2 専用の転送関数」といった分身を次々と作成し、AWS から取得した説明文（Description）を貼り付けて Kiro に提供しています。

2. 汎用プロキシ実行 (proxy_execution)

async def proxy_execution(tool_name: str, **kwargs: Any) -> str:
    # 1. 引数の皮むき (Unwrapping)
    actual_args = kwargs
    if 'kwargs' in kwargs and isinstance(kwargs['kwargs'], dict):
        actual_args = kwargs['kwargs']

    # 2. 認証トークン取得 & AWS接続
    token = fetch_access_token()
    # ...

    # 3. 指定されたツール名で実行
    result = await session.call_tool(
        name=tool_name,
        arguments=actual_args  # 皮をむいた引数を渡す
    )

Kiro がどのツールを選んでも、最終的にはこの関数が呼ばれます。 ここで「認証トークンの取得」から「引数の整形」「AWSへの転送」までを一手に引き受けます。この構造のおかげで、認証ロジックを一箇所に集約できています。

4. Kiro (IDE) への連携

最後に、この bridge.py を Kiro に認識させます。

{
  "mcpServers": {
    "fetch": {
      "command": "uvx",
      "args": ["mcp-server-fetch"],
      "env": {},
      "disabled": true,
      "autoApprove": []
    },
    "alb-logs-bridge": {
      "command": "/path-to/python3",
      "args": [
        "/path-to/bridge.py"
      ],
      "env": {
        "PYTHONUNBUFFERED": "1"
      }
    }
  }
}

kiro を再起動するとチェックが入りました。

実行結果は冒頭に紹介した通りです。

description の判定ロジックを試してみる（セマンティック・ルーティング）

Bedrock AgentCore Gateway にターゲットを二つ登録しました。 両方とも同じ Lambda をターゲットにしています。

『山本です』と明言した場合にのみ使用されるターゲット

ターゲットスキーマの description を設定し『山本です』と明言した場合にのみ使用されるターゲットを 1 つ用意しました。
get_alb_logs2 という名前にしています。

[
  {
    "description": "ユーザーが**『山本です』と明言した場合にのみ**使用してください。それ以外の場合は使用禁止です。",
    "inputSchema": {
      "properties": {
        "date": {
          "description": "検索対象の日付 (YYYY-MM-DD形式)",
          "type": "string"
        },
        "status_code": {
          "description": "HTTPステータスコード (例: 200, 500)",
          "type": "string"
        }
      },
      "required": [
        "date"
      ],
      "type": "object"
    },
    "name": "get_alb_logs2"
  }
]

kiro での実行結果：

『山本です』と明言した場合以外に使用されるターゲット

反対に、『山本です』と明言した場合以外に使用されるターゲットも 1 つ用意しました。
「ログ取得権限がないことを通知するツールです」と書いておきながら、同じ Lambda を呼び出しているので分析できてしまうのはご愛嬌で！

[
  {
    "description": "ユーザーが**『山本です』と言わなかった場合**は、必ずこちらを使用してください。ログ取得権限がないことを通知するツールです。",
    "inputSchema": {
      "properties": {
        "date": {
          "description": "検索対象の日付 (YYYY-MM-DD形式)",
          "type": "string"
        },
        "status_code": {
          "description": "HTTPステータスコード (例: 200, 500)",
          "type": "string"
        }
      },
      "required": [
        "date"
      ],
      "type": "object"
    },
    "name": "get_alb_logs"
  }
]

kiro での実行結果：

セマンティック・ルーティングが機能していますね。

まとめ

  • Bedrock AgentCore Gateway は、Lambda を MCP ツールとして公開するのに非常に便利。
  • Kiro などの AI IDE は MCP を標準サポートしているため、自作ツールを簡単に組み込める。
  • これらを組み合わせると、開発体験（DevEx）が劇的に向上する。

今回は「ログ解析」でしたが、これを応用すれば「EC2の再起動」や「DBのデータ修正」もチャットだけで完結できそうです。これぞ ChatOps の未来 ですね。 と言いつつ bridge.py の作成はかなり苦労しました。この辺の連携が楽になるといいですが・・という課題を残して、この記事を終えようと思います。 kiro はフリープランで 50 クレジットまで使えるので、一旦それでここまで検証しています。

山本 哲也 (記事一覧)

カスタマーサクセス部のインフラエンジニア。

山を走るのが趣味です。