Amazon Bedrock Knowledge Base MCP Server

Instructions

Implementation Reference

• src/bedrock_kb_mcp_server/main.py:807-848 (handler)

  The 'get_ingestion_job' tool is registered with @mcp.tool() and handles the ingestion job retrieval, calling the bedrock_client.

  @mcp.tool()  # MCPツールとして公開
  @handle_errors  # エラーハンドリングデコレータを適用
  def get_ingestion_job(
      knowledge_base_id: str, data_source_id: str, ingestion_job_id: str
  ) -> IngestionJobResponseDict:
      """
      取り込みジョブのステータスと詳細情報を取得します。
      
      取り込みジョブの進捗状況、統計情報、エラー情報などを取得できます。
  
      Args:
          knowledge_base_id: Knowledge BaseのID
          data_source_id: データソースのID
          ingestion_job_id: 取り込みジョブのID（`start_ingestion_job`で取得）
  
      Returns:
          IngestionJobResponseDict: 取り込みジョブの詳細情報
              - ingestion_job_id: 取り込みジョブのID
              - status: ジョブのステータス
                  - "STARTING": ジョブが開始中
                  - "IN_PROGRESS": ジョブが実行中
                  - "COMPLETE": ジョブが完了
                  - "FAILED": ジョブが失敗
              - statistics: 統計情報（オプション、ジョブが進行中または完了している場合）
                  - numberOfDocumentsScanned: スキャンされたドキュメント数
                  - numberOfDocumentsModified: 変更されたドキュメント数
                  - numberOfDocumentsDeleted: 削除されたドキュメント数
                  - numberOfDocumentsFailed: 失敗したドキュメント数
      
      Raises:
          ValueError: いずれかのIDが空の場合
      """
      # 入力値のバリデーション（共通関数を使用）
      knowledge_base_id = validate_required_string(knowledge_base_id, "knowledge_base_id")
      data_source_id = validate_required_string(data_source_id, "data_source_id")
      ingestion_job_id = validate_required_string(ingestion_job_id, "ingestion_job_id")
      
      # Bedrockクライアントから取り込みジョブの詳細を取得
      result = bedrock_client.get_ingestion_job(
          knowledge_base_id, data_source_id, ingestion_job_id
      )
      return result

• src/bedrock_kb_mcp_server/bedrock_client.py:486-520 (handler)

  The underlying implementation of 'get_ingestion_job' that makes the AWS Bedrock API call via bedrock_agent.

  def get_ingestion_job(
      self, knowledge_base_id: str, data_source_id: str, ingestion_job_id: str
  ) -> IngestionJobResponseDict:
      """
      取り込みジョブのステータスと詳細情報を取得します。
      
      取り込みジョブの進捗状況、統計情報、エラー情報などを取得できます。
      ジョブの完了を待つために、この関数を定期的に呼び出すことができます。
  
      Args:
          knowledge_base_id: Knowledge BaseのID
          data_source_id: データソースのID
          ingestion_job_id: 取り込みジョブのID（`start_ingestion_job`で取得）
  
      Returns:
          IngestionJobResponseDict: 取り込みジョブの詳細情報
              - ingestion_job_id: 取り込みジョブのID
              - status: ジョブのステータス
                  - "STARTING": ジョブが開始中
                  - "IN_PROGRESS": ジョブが実行中
                  - "COMPLETE": ジョブが完了
                  - "FAILED": ジョブが失敗
              - statistics: 統計情報（オプション、ジョブが進行中または完了している場合）
                  - numberOfDocumentsScanned: スキャンされたドキュメント数
                  - numberOfDocumentsModified: 変更されたドキュメント数
                  - numberOfDocumentsDeleted: 削除されたドキュメント数
                  - numberOfDocumentsFailed: 失敗したドキュメント数
      
      Raises:
          ClientError: AWS API呼び出しが失敗した場合
      """
      try:
          # AWS Bedrock APIを呼び出して取り込みジョブの詳細を取得
          response = self.bedrock_agent.get_ingestion_job(
              knowledgeBaseId=knowledge_base_id,