list_s3_documents
Retrieve a list of documents from an Amazon S3 bucket, optionally filtering by folder prefix to locate specific files for knowledge base management.
Instructions
S3バケット内のドキュメント一覧を取得します。
指定されたプレフィックス（フォルダ）に一致するドキュメントのみを 取得することもできます。
Args: bucket_name: S3バケット名 prefix: フィルタリングするS3プレフィックス（オプション） 例: "documents/" を指定すると、documents/フォルダ内の ファイルのみが返されます
Returns: S3DocumentListResponseDict: ドキュメント一覧 - count: ドキュメントの数 - bucket: バケット名 - prefix: 使用されたプレフィックス（指定した場合） - documents: ドキュメントの詳細情報のリスト 各要素には以下の情報が含まれます: - key: S3オブジェクトキー（ファイルパス） - size: ファイルサイズ（バイト） - last_modified: 最終更新日時（ISO形式）
Raises: ValueError: bucket_nameが空の場合
Example: # すべてのドキュメントを取得 list_s3_documents("my-bucket")
# 特定のプレフィックスのドキュメントのみを取得
list_s3_documents("my-bucket", "documents/")Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bucket_name | Yes | ||
| prefix | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| count | Yes | ||
| bucket | Yes | ||
| prefix | Yes | ||
| documents | Yes |
Implementation Reference
- Implementation of the S3 document listing logic using the S3 client paginator.
def list_s3_documents( self, bucket_name: str, prefix: str = "" ) -> List[Dict[str, Any]]: """ S3バケット内のドキュメント一覧を取得します。 ページネーションを使用して、すべてのドキュメントを取得します。 指定されたプレフィックス（フォルダ）に一致するドキュメントのみを 取得することもできます。 Args: bucket_name: S3バケット名 prefix: フィルタリングするS3プレフィックス（オプション） 例: "documents/" を指定すると、documents/フォルダ内の ファイルのみが返されます Returns: List[Dict[str, Any]]: ドキュメントの詳細情報のリスト 各要素には以下の情報が含まれます: - key: S3オブジェクトキー（ファイルパス） - size: ファイルサイズ（バイト） - last_modified: 最終更新日時（ISO形式） Raises: ClientError: AWS API呼び出しが失敗した場合 例: バケットが存在しない、権限がないなど Note: 大量のドキュメントがある場合、この関数の実行に時間がかかる場合があります。 """ try: documents = [] # ページネーターを取得（複数ページの結果を自動的に処理） paginator = self.s3_client.get_paginator("list_objects_v2") # すべてのページをループして結果を収集 for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix): # 各ページからオブジェクト情報を取得 for obj in page.get("Contents", []): # ドキュメント情報を整形してリストに追加 documents.append( { "key": obj["Key"], # S3オブジェクトキー "size": obj["Size"], # ファイルサイズ（バイト） "last_modified": obj["LastModified"].isoformat(), # ISO形式の日時 } ) # 取得したドキュメントの数をログに記録 logger.info(f"Retrieved {len(documents)} documents from S3") return documents except ClientError as e: logger.error(f"Error listing S3 documents: {e}") raise - src/bedrock_kb_mcp_server/main.py:964-1016 (registration)MCP tool registration and wrapper function for list_s3_documents.
@mcp.tool() # MCPツールとして公開 @handle_errors # エラーハンドリングデコレータを適用 def list_s3_documents(bucket_name: str, prefix: str = "") -> S3DocumentListResponseDict: """ S3バケット内のドキュメント一覧を取得します。 指定されたプレフィックス（フォルダ）に一致するドキュメントのみを 取得することもできます。 Args: bucket_name: S3バケット名 prefix: フィルタリングするS3プレフィックス（オプション） 例: "documents/" を指定すると、documents/フォルダ内の ファイルのみが返されます Returns: S3DocumentListResponseDict: ドキュメント一覧 - count: ドキュメントの数 - bucket: バケット名 - prefix: 使用されたプレフィックス（指定した場合） - documents: ドキュメントの詳細情報のリスト 各要素には以下の情報が含まれます: - key: S3オブジェクトキー（ファイルパス） - size: ファイルサイズ（バイト） - last_modified: 最終更新日時（ISO形式） Raises: ValueError: bucket_nameが空の場合 Example: # すべてのドキュメントを取得 list_s3_documents("my-bucket") # 特定のプレフィックスのドキュメントのみを取得 list_s3_documents("my-bucket", "documents/") """ # 入力値のバリデーション（共通関数を使用） bucket_name = validate_required_string(bucket_name, "bucket_name") # プレフィックスはオプションなので、指定されている場合のみstripを適用 prefix_cleaned = prefix.strip() if prefix else "" # BedrockクライアントからS3ドキュメント一覧を取得 documents = bedrock_client.list_s3_documents( bucket_name, # 前後の空白は既に削除済み prefix_cleaned ) return { "count": len(documents), "bucket": bucket_name, # 前後の空白は既に削除済み "prefix": prefix_cleaned, "documents": documents, }