Amazon Bedrock Knowledge Base MCP Server

Instructions

Implementation Reference

• src/bedrock_kb_mcp_server/main.py:767-804 (handler)

  Tool registration and handler for start_ingestion_job in main.py

  @mcp.tool()  # MCPツールとして公開
  @handle_errors  # エラーハンドリングデコレータを適用
  def start_ingestion_job(
      knowledge_base_id: str, data_source_id: str
  ) -> IngestionJobResponseDict:
      """
      データソースからKnowledge Baseへのデータ取り込みジョブを開始します。
      
      このジョブは非同期で実行され、データソース内のドキュメントを
      Knowledge Baseに取り込みます。ジョブの進捗は`get_ingestion_job`で確認できます。
  
      Args:
          knowledge_base_id: Knowledge BaseのID
          data_source_id: データソースのID
  
      Returns:
          IngestionJobResponseDict: 取り込みジョブの開始結果
              - ingestion_job_id: 開始された取り込みジョブのID
              - status: ジョブのステータス（通常は "STARTING" または "IN_PROGRESS"）
              - statistics: 統計情報（オプション、ジョブ開始時は通常None）
      
      Raises:
          ValueError: knowledge_base_idまたはdata_source_idが空の場合
      
      Note:
          取り込みジョブは非同期で実行されるため、この関数は即座に返ります。
          ジョブの完了を待つには、`get_ingestion_job`を定期的に呼び出して
          ステータスを確認してください。
      """
      # 入力値のバリデーション（共通関数を使用）
      knowledge_base_id = validate_required_string(knowledge_base_id, "knowledge_base_id")
      data_source_id = validate_required_string(data_source_id, "data_source_id")
      
      # Bedrockクライアントを使用して取り込みジョブを開始
      result = bedrock_client.start_ingestion_job(
          knowledge_base_id, data_source_id
      )
      return result

• src/bedrock_kb_mcp_server/bedrock_client.py:439-484 (handler)

  Actual Bedrock API implementation of start_ingestion_job in BedrockKBClient class

  def start_ingestion_job(
      self, knowledge_base_id: str, data_source_id: str
  ) -> IngestionJobResponseDict:
      """
      データソースからKnowledge Baseへのデータ取り込みジョブを開始します。
      
      このジョブは非同期で実行され、データソース内のドキュメントを
      Knowledge Baseに取り込みます。取り込みには時間がかかる場合があります。
  
      Args:
          knowledge_base_id: Knowledge BaseのID
          data_source_id: データソースのID
  
      Returns:
          IngestionJobResponseDict: 取り込みジョブの開始結果
              - ingestion_job_id: 開始された取り込みジョブのID
              - status: ジョブのステータス（通常は "STARTING" または "IN_PROGRESS"）
              - statistics: 統計情報（オプション、ジョブ開始時は通常None）
      
      Raises:
          ClientError: AWS API呼び出しが失敗した場合
      
      Note:
          取り込みジョブは非同期で実行されるため、この関数は即座に返ります。
          ジョブの進捗を確認するには、`get_ingestion_job`を使用してください。
      """
      try:
          # AWS Bedrock APIを呼び出して取り込みジョブを開始
          response = self.bedrock_agent.start_ingestion_job(
              knowledgeBaseId=knowledge_base_id, dataSourceId=data_source_id
          )
          
          # ジョブ開始成功をログに記録
          logger.info(
              f"Started ingestion job {response['ingestionJob']['ingestionJobId']} "
              f"for data source {data_source_id}"
          )
          
          # レスポンスを整形して返す
          return {
              "ingestion_job_id": response["ingestionJob"]["ingestionJobId"],
              "status": response["ingestionJob"]["status"],
          }
      except ClientError as e:
          logger.error(f"Error starting ingestion job: {e}")
          raise