Amazon Bedrock Knowledge Base MCP Server (Unofficial)

Amazon Bedrock Knowledge Baseを管理するためのMCP（Model Context Protocol）サーバーです。

本プロジェクトでは FastMCPフレームワークを使用して Bedrock Knowledge Base に対する操作と RAG（Retrieval-Augmented Generation）機能を提供しています。

リポジトリ

https://github.com/r3-yamauchi/bedrock-kb-mcp-server

起動方法

主な機能

• Knowledge Base管理: 作成、一覧取得、詳細取得、更新（敢えて削除機能は実装していません）

  • ストレージタイプは S3 のみをサポート

  • カスタムパーシング設定とチャンキング設定のサポート

• データソース管理: 作成、一覧取得（敢えて削除機能は実装していません）

  • カスタムパーシング設定とチャンキング設定のサポート

• データ取り込みジョブ: 開始、ステータス確認

• RAGクエリ: Knowledge Baseに対する検索クエリの実行

• S3ドキュメント管理: アップロード、一覧取得

プロジェクト構造

技術スタック

コアライブラリ

• FastMCP (>=0.1.0): MCPサーバーを構築するためのフレームワーク

• boto3 (>=1.26.0): AWS SDK for Python - AWSサービスとの通信に使用

• pydantic (>=2.0.0): データバリデーションと設定管理

開発ツール

• pytest (>=7.0): テストフレームワーク

• pytest-asyncio (>=0.21.0): 非同期テストサポート

• black (>=23.0): コードフォーマッター

• ruff (>=0.1.0): 高速なPythonリンター

• mypy (>=1.0): 静的型チェッカー

Python要件

• Python 3.12以上が必要です

クイックスタート

前提条件

• Python 3.12以上がインストールされていること

• uvがインストールされていること

• AWSアカウントと適切な認証情報が設定されていること

1. 依存関係のインストール

プロジェクトのルートディレクトリで以下のコマンドを実行します：

開発環境の場合は：

2. AWS認証情報の設定

以下のいずれかの方法でAWS認証情報を設定します：

方法1: AWSプロファイルを使用（推奨）

方法2: 環境変数で直接設定

方法3: AWS CLIで設定済みの場合

AWS CLIでaws configureを実行して設定済みの場合は、追加の設定は不要です。

3. 環境変数の設定（オプション）

ログレベルや構造化ログの設定を行います：

4. サーバーの起動

以下のコマンドでサーバーを起動します：

正常に起動すると、以下のようなログが表示されます：

5. 動作確認

MCP Serverは標準入力（stdin）からJSON-RPC形式のリクエストを受け取り、標準出力（stdout）にJSON-RPC形式のレスポンスを返します。

動作確認方法1: サーバーが起動していることを確認

サーバーを起動すると、以下のようなログが表示されます。このログが表示されれば、サーバーは正常に起動しています。

動作確認方法2: MCPクライアントを使用（推奨）

MCP対応のクライアント（例: Claude Desktop、Cursor IDEなど）を使用して接続します。

Claude Desktopの場合:

1. ~/Library/Application Support/Claude/claude_desktop_config.jsonを編集

2. MCP Serverを追加：

Cursor IDEの場合:

1. 設定からMCP Serverを追加

2. コマンドパスと環境変数を設定

動作確認方法3: テストスクリプトを使用

プロジェクトに含まれているtest_mcp_server.pyを使用：

注意: このスクリプトはサーバーが起動してリクエストに応答することを確認しますが、実際のAWS API呼び出しは行いません。

インストール

開発環境の場合は：

使用方法

環境変数の設定

以下の環境変数を設定する必要があります：

• AWS_PROFILE: AWSプロファイル名（認証情報の管理に使用）

• AWS_REGION: AWSリージョン（例: us-east-1、デフォルト: us-east-1）

• FASTMCP_LOG_LEVEL: ログレベル（DEBUG, INFO, WARNING, ERROR, CRITICAL、デフォルト: INFO）

• FASTMCP_STRUCTURED_LOG: 構造化ログ（JSON形式）を使用するか（true/false、デフォルト: false）

サーバーの起動

主要コンポーネント

1. bedrock_client.py - AWS Bedrock APIクライアント

AWS Bedrock Knowledge Base APIとの低レベルな通信を担当するラッパークラスです。

BedrockKBClient クラス

環境変数AWS_REGIONからリージョンを取得し、以下の3つのAWSクライアントを初期化します：

• bedrock-agent: Knowledge Baseとデータソースの管理用

• bedrock-agent-runtime: RAGクエリ実行用

• s3: S3ドキュメント管理用

すべてのクライアントには以下の設定が適用されます：

• リトライ設定（最大3回、adaptiveモード）

• 接続タイムアウト（10秒）

• 読み取りタイムアウト（30秒）

主要メソッド

Knowledge Base管理

• create_knowledge_base(): 新しいKnowledge Baseを作成

• list_knowledge_bases(): すべてのKnowledge Baseを一覧取得（ページネーション対応）

• get_knowledge_base(): 特定のKnowledge Baseの詳細情報を取得

• update_knowledge_base(): Knowledge Baseの名前、説明、IAMロールを更新

データソース管理

• create_data_source(): Knowledge Baseにデータソースを追加

• list_data_sources(): 指定されたKnowledge Baseのデータソース一覧を取得

データ取り込みジョブ管理

• start_ingestion_job(): データソースからKnowledge Baseへのデータ取り込みジョブを開始

• get_ingestion_job(): 取り込みジョブのステータスと統計情報を取得

RAGクエリ

• retrieve(): Knowledge Baseに対してRAGクエリを実行（結果数1-100を指定可能）

S3ドキュメント管理

• upload_document_to_s3(): ローカルファイルをS3バケットにアップロード

• list_s3_documents(): S3バケット内のドキュメント一覧を取得（プレフィックスでフィルタリング可能）

2. main.py - MCPサーバーメイン

FastMCPフレームワークを使用してMCPサーバーを構築し、BedrockKBClientの機能をMCPツールとして公開します。

MCPツール

Knowledge Base管理ツール

• create_knowledge_base: Knowledge Baseを作成

  • ストレージタイプ: S3、S3_VECTORS

  • 埋め込みモデル: Amazon Titan、Cohere、Amazon Nova Multimodal Embeddings v1

  • パーシング設定: BEDROCK_FOUNDATION_MODEL、BEDROCK_DATA_AUTOMATION

  • チャンキング設定: FIXED_SIZE、HIERARCHICAL、SEMANTIC、NONE

  • マルチモーダルストレージ設定（supplementalDataStorageConfiguration）

  • S3 ARN形式とS3 URI形式の両方をサポート

  • IAMロールARNのアカウントID自動補完

• list_knowledge_bases: すべてのKnowledge Baseを一覧取得

• get_knowledge_base: 特定のKnowledge Baseの詳細を取得

• update_knowledge_base: Knowledge Baseを更新

データソース管理ツール

• create_data_source: データソースを作成

  • パーシング設定とチャンキング設定のサポート

  • S3 ARN形式とS3 URI形式の両方をサポート

• list_data_sources: データソース一覧を取得

データ取り込みツール

• start_ingestion_job: 取り込みジョブを開始

• get_ingestion_job: 取り込みジョブのステータスを取得

RAGクエリツール

• retrieve: Knowledge Baseに対してRAGクエリを実行

S3ドキュメント管理ツール

• upload_document_to_s3: S3にドキュメントをアップロード

• list_s3_documents: S3バケット内のドキュメント一覧を取得

3. models.py - Pydanticモデル

リクエスト/レスポンスのバリデーションと型安全性を提供するPydanticモデルを定義します。

• StorageType: ストレージタイプの列挙型（S3, S3_VECTORS）

• SourceType: データソースタイプの列挙型（S3）

• ParsingStrategy: パーシング戦略の列挙型（BEDROCK_FOUNDATION_MODEL, BEDROCK_DATA_AUTOMATION）

• ChunkingStrategy: チャンキング戦略の列挙型（FIXED_SIZE, HIERARCHICAL, SEMANTIC, NONE）

• ParsingConfiguration: パーシング設定モデル

• ChunkingConfiguration: チャンキング設定モデル

• VectorIngestionConfiguration: ベクトル取り込み設定モデル

• CreateKnowledgeBaseRequest: Knowledge Base作成リクエストのバリデーション

  • S3 URI形式のサポート（自動的にARN形式に変換）

  • IAMロールARNのアカウントID自動補完

• CreateDataSourceRequest: データソース作成リクエストのバリデーション

  • S3 URI形式のサポート（自動的にARN形式に変換）

• 各種レスポンスモデル

4. types.py - TypedDict定義

APIレスポンスの型安全性を向上させるためのTypedDict定義を提供します。

• KnowledgeBaseResponseDict: Knowledge Base作成/更新レスポンス

• DataSourceResponseDict: データソース作成レスポンス

• IngestionJobResponseDict: 取り込みジョブレスポンス

• その他のレスポンス型定義

5. utils.py - ユーティリティ関数

設定管理、エラーハンドリング、ログ出力、ARN正規化などの共通機能を提供します。

• validate_aws_credentials(): AWS認証情報の検証

• get_log_level(): ログレベルの安全な取得

• handle_errors(): エラーハンドリングデコレータ（AWS APIエラーの適切な処理）

  • 10種類以上のAWSエラーコードに対応

  • AWSリクエストIDを含む詳細なエラー情報

• get_aws_account_id(): STSを使用してAWSアカウントIDを取得

• normalize_s3_arn_or_uri(): S3 URI形式をARN形式に変換

• normalize_iam_role_arn(): IAMロールARNのアカウントIDを自動補完

• validate_required_string(): 必須文字列パラメータのバリデーション共通化

• StructuredFormatter: 構造化ログフォーマッター（JSON形式）

• sanitize_log_data(): 機密情報のマスキング

• setup_logging(): ロギング設定の一元管理

ワークフロー例

1. Knowledge Baseの作成と設定

1. S3バケットにドキュメントをアップロード

   upload_document_to_s3(local_file_path, bucket_name, s3_key)

2. Knowledge Baseを作成（S3 URI形式とIAMロールARNの短縮形式を使用可能）

   # 基本的なKnowledge Base create_knowledge_base( name="My Knowledge Base", description="Example KB", role_arn="role/BedrockKBRole", # アカウントIDなし形式（自動補完） storage_type="S3", bucket_arn="s3://my-bucket" # S3 URI形式 ) # S3 Vectorsを使用したKnowledge Base create_knowledge_base( name="Vector KB", description="Vector search enabled KB", role_arn="arn:aws:iam::123456789012:role/BedrockKBRole", storage_type="S3_VECTORS", bucket_arn="s3://vector-bucket", embedding_model_arn="arn:aws:bedrock:us-east-1::foundation-model/amazon.titan-embed-text-v1" ) # マルチモーダルKnowledge Base（Amazon Nova Multimodal Embeddings v1） create_knowledge_base( name="Multimodal KB", description="KB with Nova Multimodal Embeddings", role_arn="role/BedrockKBRole", storage_type="S3_VECTORS", bucket_arn="s3://vector-bucket", embedding_model_arn="arn:aws:bedrock:us-east-1::foundation-model/amazon.nova-2-multimodal-embeddings-v1:0", multimodal_storage_s3_uri="s3://multimodal-storage-bucket/" ) # カスタムパーシングとチャンキング設定を使用 create_knowledge_base( name="Custom KB", description="KB with custom parsing and chunking", role_arn="role/BedrockKBRole", storage_type="S3", bucket_arn="s3://my-bucket", parsing_strategy="BEDROCK_FOUNDATION_MODEL", parsing_model_arn="arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0", parsing_modality="MULTIMODAL", chunking_strategy="FIXED_SIZE", chunking_max_tokens=1000, chunking_overlap_percentage=20 )

3. データソースを作成（S3 URI形式とカスタム設定を使用可能）

   # 基本的なデータソース create_data_source( knowledge_base_id="KB123", name="My Data Source", source_type="S3", bucket_arn="s3://my-bucket" # S3 URI形式 ) # カスタムパーシングとチャンキング設定を使用 create_data_source( knowledge_base_id="KB123", name="Custom Data Source", source_type="S3", bucket_arn="s3://my-bucket", parsing_strategy="BEDROCK_FOUNDATION_MODEL", parsing_model_arn="arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-3-sonnet-20240229-v1:0", chunking_strategy="FIXED_SIZE", chunking_max_tokens=1000, chunking_overlap_percentage=20 )

4. データ取り込みジョブを開始

   start_ingestion_job(knowledge_base_id, data_source_id)

5. ジョブのステータスを確認

   get_ingestion_job(knowledge_base_id, data_source_id, ingestion_job_id)

2. RAGクエリの実行

1. Knowledge Baseに対してクエリを実行

   retrieve(knowledge_base_id, query, number_of_results)

2. 結果には関連ドキュメントと引用情報が含まれます

アーキテクチャの特徴

1. レイヤードアーキテクチャ

• プレゼンテーション層: main.py - MCPツールの定義と公開

• ビジネスロジック層: bedrock_client.py - AWS APIとの通信

• データ層: models.py - データバリデーションと型定義

2. エラーハンドリング

• すべてのAWS API呼び出しはClientErrorを適切に処理

• 10種類以上のAWSエラーコードに対応（InternalServerException、InvalidParameterException、ResourceNotFoundExceptionなど）

• AWSリクエストIDを含む詳細なエラー情報

• エラーコードに応じた日本語メッセージを提供

• エラー情報はログに記録され、ユーザーには分かりやすいメッセージが返却される

3. ロギング

• 標準ログ: 人間が読みやすい形式（デフォルト）

• 構造化ログ: JSON形式で出力可能（FASTMCP_STRUCTURED_LOG=trueで有効化）

• 機密情報の自動マスキング: ARN、認証情報などが自動的にマスクされる

• 環境変数でログレベルを制御可能

4. 型安全性

• Pydanticを使用したデータバリデーション

• TypedDictを使用したAPIレスポンスの型定義

• すべてのMCPツール関数とBedrockKBClientメソッドの戻り値型を具体的に定義

• 型ヒントの活用

• mypyによる静的型チェック対応

• IDEの補完機能と型安全性が向上

5. 入力バリデーションと正規化

• ARN形式とURI形式の両方をサポート

  • S3 ARN形式: arn:aws:s3:::bucket-name

  • S3 URI形式: s3://bucket-name または s3://bucket-name/path（自動的にARN形式に変換）

  • IAMロールARN: 完全な形式または短縮形式（アカウントIDを自動補完）

• 必須フィールドのチェック

• ストレージタイプ、パーシング戦略、チャンキング戦略の列挙型による型安全性の向上

• Pydanticモデルによる包括的なバリデーション

6. リトライロジック

• AWS API呼び出しにリトライメカニズムを実装

• 適応的リトライモード（adaptive）を使用

• 一時的なネットワークエラーやレート制限エラーに対する自動リトライ

セキュリティ考慮事項

1. AWS認証情報: 環境変数AWS_PROFILEを使用して認証情報を管理

2. IAMロール: Knowledge Baseには適切なIAMロールが必要

3. S3アクセス: S3バケットへのアクセス権限が必要

4. リージョン設定: 適切なAWSリージョンを指定

5. 機密情報のマスキング: ログ出力時にARN、認証情報などが自動的にマスクされる

トラブルシューティング

よくある問題

1. AWS認証エラー

   • AWS_PROFILEが正しく設定されているか確認

   • AWS認証情報が有効か確認

   • boto3が自動的に認証情報を取得する場合もあるため（IAMロール、EC2インスタンスプロファイル）、明示的な設定がなくても動作する場合があります

2. リージョンエラー

   • AWS_REGIONが正しく設定されているか確認

   • Bedrockがそのリージョンで利用可能か確認

3. 権限エラー

   • IAMロールに必要な権限があるか確認

   • S3バケットへのアクセス権限があるか確認

4. 取り込みジョブの失敗

   • データソースの設定を確認

   • S3バケット内のドキュメント形式を確認

5. バリデーションエラー

   • ARN形式が正しい場合、エラーメッセージを確認

   • 必須フィールドがすべて指定されているか確認

エラー: AWS認証情報が見つからない

解決方法: AWS認証情報を設定してください（クイックスタートの手順2を参照）

エラー: モジュールが見つからない

解決方法: 依存関係をインストールしてください

エラー: Pythonバージョンが古い

解決方法: Python 3.12以上をインストールしてください

サーバーが起動しない

• ログレベルをDEBUGに設定して詳細なログを確認：

• 構造化ログを有効にしてJSON形式で確認：

実際の使用例

Knowledge Baseの一覧を取得

MCPクライアントからlist_knowledge_basesツールを呼び出すと、AWSアカウント内のすべてのKnowledge Baseが返されます。

Knowledge Baseを作成

注意:

• role_arnは短縮形式（role/ROLE_NAME）も使用可能で、アカウントIDが自動補完されます

• bucket_arnはS3 URI形式（s3://bucket-name）も使用可能で、自動的にARN形式に変換されます

RAGクエリを実行

注意事項

• 実際のAWS APIを呼び出すため、適切なAWS認証情報とIAM権限が必要です

• Knowledge Baseの作成には、適切なIAMロールが必要です

• S3バケットへのアクセス権限が必要です

• リージョンによってはBedrockが利用できない場合があります

開発

コードフォーマット

テスト（今後実装予定）

ライセンス

MIT License - see LICENSE file for details.