Amazon Bedrock Knowledge Base MCP Server

Instructions

Implementation Reference

• src/bedrock_kb_mcp_server/bedrock_client.py:603-641 (handler)

  The core handler logic that executes the S3 file upload using boto3.

  def upload_document_to_s3(
      self, local_file_path: str, bucket_name: str, s3_key: str
  ) -> S3UploadResponseDict:
      """
      ローカルファイルをS3バケットにアップロードします。
      
      アップロードされたファイルは、Knowledge Baseのデータソースとして
      使用できます。
  
      Args:
          local_file_path: アップロードするローカルファイルのパス
          bucket_name: アップロード先のS3バケット名
          s3_key: S3オブジェクトキー（バケット内のパス）
                  例: "documents/myfile.pdf" のようにパスを指定可能
  
      Returns:
          S3UploadResponseDict: アップロード結果
              - s3_uri: アップロードされたファイルのS3 URI（s3://bucket/key形式）
              - status: アップロードステータス（"uploaded"）
      
      Raises:
          ClientError: AWS API呼び出しが失敗した場合
              例: バケットが存在しない、権限がない、ファイルが大きすぎるなど
      """
      try:
          # S3クライアントを使用してファイルをアップロード
          self.s3_client.upload_file(local_file_path, bucket_name, s3_key)
          
          # S3 URIを構築
          s3_uri = f"s3://{bucket_name}/{s3_key}"
          
          # アップロード成功をログに記録
          logger.info(f"Uploaded document to {s3_uri}")
          
          # アップロード結果を返す
          return {"s3_uri": s3_uri, "status": "uploaded"}
      except ClientError as e:
          logger.error(f"Error uploading document to S3: {e}")
          raise

• src/bedrock_kb_mcp_server/main.py:907-961 (registration)

  The MCP tool registration and input validation wrapper for 'upload_document_to_s3'.

  def upload_document_to_s3(
      local_file_path: str, bucket_name: str, s3_key: str
  ) -> S3UploadResponseDict:
      """
      ローカルファイルをS3バケットにアップロードします。
      
      アップロードされたファイルは、Knowledge Baseのデータソースとして
      使用できます。
  
      Args:
          local_file_path: アップロードするローカルファイルのパス
          bucket_name: アップロード先のS3バケット名
          s3_key: S3オブジェクトキー（バケット内のパス）
                  例: "documents/myfile.pdf" のようにパスを指定可能
  
      Returns:
          S3UploadResponseDict: アップロード結果
              - s3_uri: アップロードされたファイルのS3 URI（s3://bucket/key形式）
              - status: アップロードステータス（"uploaded"）
      
      Raises:
          ValueError: パラメータが空の場合、またはファイルが存在しない場合
      
      Example:
          upload_document_to_s3(
              "/path/to/document.pdf",
              "my-bucket",
              "documents/document.pdf"
          )
          # 戻り値: {"s3_uri": "s3://my-bucket/documents/document.pdf", "status": "uploaded"}
      """
      import os
      
      # 入力値のバリデーション（共通関数を使用）
      # すべてのパラメータは必須です
      local_file_path = validate_required_string(local_file_path, "local_file_path")
      bucket_name = validate_required_string(bucket_name, "bucket_name")
      s3_key = validate_required_string(s3_key, "s3_key")
      
      # ファイルの存在確認
      # ローカルファイルシステム上にファイルが存在することを確認します
      # ファイルが存在しない場合、S3へのアップロードは失敗するため、事前にチェックします
      if not os.path.exists(local_file_path):
          raise ValueError(f"File not found: {local_file_path}")
      
      # 注意: ファイルサイズのチェックは行っていません
      # 非常に大きなファイルの場合、アップロードに時間がかかる可能性があります
  
      # Bedrockクライアントを使用してS3にアップロード
      result = bedrock_client.upload_document_to_s3(
          local_file_path,  # 前後の空白は既に削除済み
          bucket_name,  # 前後の空白は既に削除済み
          s3_key  # 前後の空白は既に削除済み
      )
      return result