Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/alerts.py:947-952 (registration)

  The @mcp_tool decorator that registers the 'start_ai_alert_triage' tool, specifying permissions (RUN_PANTHER_AI) and readOnlyHint.

  @mcp_tool(
      annotations={
          "permissions": all_perms(Permission.RUN_PANTHER_AI),
          "readOnlyHint": True,
      }
  )

• src/mcp_panther/panther_mcp_core/tools/alerts.py:953-973 (schema)

  Input schema defined via Annotated Pydantic fields: alert_id (str, required), prompt (str | None, optional), timeout_seconds (int, default=180, ge=120 le=300).

  async def start_ai_alert_triage(
      alert_id: Annotated[
          str,
          Field(min_length=1, description="The ID of the alert to start AI triage for"),
      ],
      prompt: Annotated[
          str | None,
          Field(
              min_length=1,
              description="Optional additional prompt to provide context for the AI triage",
          ),
      ] = None,
      timeout_seconds: Annotated[
          int,
          Field(
              description="Timeout in seconds to wait for AI triage completion",
              ge=120,
              le=300,
          ),
      ] = 180,
  ) -> dict[str, Any]:

• src/mcp_panther/panther_mcp_core/tools/alerts.py:974-1110 (handler)

  The core handler logic: constructs GraphQL variables for AI_SUMMARIZE_ALERT_MUTATION to start triage stream, polls AI_INFERENCE_STREAM_QUERY until finished or timeout, accumulates responseText, handles errors and returns summary with metadata.

  """Start an AI-powered triage analysis for a Panther alert with intelligent insights and recommendations.
  
  This tool initiates Panther's embedded AI agent to triage an alert and provide
  an intelligent report about the events, risk level, potential impact, and
  recommended next steps for investigation.
  
  The AI triage includes analysis of:
  - Alert metadata (severity, detection rule, timestamps)
  - Related events and logs (if available)
  - Comments from previous investigations
  - Contextual security analysis and recommendations
  
  Returns:
      Dict containing:
      - success: Boolean indicating if triage was generated successfully
      - summary: The AI-generated triage summary text
      - stream_id: The stream ID used for this analysis
      - metadata: Information about the analysis request
      - message: Error message if unsuccessful
  """
  logger.info(f"Starting AI triage for alert {alert_id}")
  
  try:
      # Set output length to medium (fixed, not configurable by AI)
      output_length = "medium"
  
      # Prepare the AI summarize request with minimal required fields
      request_input = {
          "alertId": alert_id,
          "outputLength": output_length,
          "metadata": {
              "kind": "alert"  # Required: tells AI this is an alert analysis
          },
      }
  
      # Add optional prompt if provided
      if prompt:
          request_input["prompt"] = prompt
  
      variables = {"input": request_input}
  
      if prompt:
          logger.info(f"Using additional prompt: {prompt[:100]}...")
  
      logger.info(f"Initiating AI triage with output_length={output_length}")
  
      # Step 1: Start the AI triage
      result = await _execute_query(AI_SUMMARIZE_ALERT_MUTATION, variables)
  
      if not result or "aiSummarizeAlert" not in result:
          logger.error("Failed to initiate AI triage")
          return {
              "success": False,
              "message": "Failed to initiate AI triage",
          }
  
      stream_id = result["aiSummarizeAlert"]["streamId"]
      logger.info(f"AI triage started with stream ID: {stream_id}")
  
      # Step 2: Poll for results with timeout
      start_time = asyncio.get_event_loop().time()
      poll_interval = 2.0  # Start with 2 second intervals
      max_poll_interval = 10.0  # Maximum 10 second intervals
  
      accumulated_response = ""
  
      while True:
          current_time = asyncio.get_event_loop().time()
          elapsed = current_time - start_time
  
          if elapsed > timeout_seconds:
              logger.warning(f"AI triage timed out after {timeout_seconds} seconds")
              return {
                  "success": False,
                  "message": f"AI triage generation timed out after {timeout_seconds} seconds",
                  "stream_id": stream_id,
                  "partial_summary": accumulated_response
                  if accumulated_response
                  else None,
              }
  
          # Poll the inference stream
          stream_variables = {"streamId": stream_id}
          stream_result = await _execute_query(
              AI_INFERENCE_STREAM_QUERY, stream_variables
          )
  
          if not stream_result or "aiInferenceStream" not in stream_result:
              logger.error("Failed to poll AI inference stream")
              await asyncio.sleep(poll_interval)
              continue
  
          inference_data = stream_result["aiInferenceStream"]
  
          # Check for errors
          if inference_data.get("error"):
              logger.error(f"AI inference error: {inference_data['error']}")
              return {
                  "success": False,
                  "message": f"AI inference failed: {inference_data['error']}",
                  "stream_id": stream_id,
              }
  
          # Get the latest response text (AI streams the full response each time)
          response_text = inference_data.get("responseText", "")
          if response_text:
              accumulated_response = response_text
  
          # Check if finished
          if inference_data.get("finished", False):
              logger.info(f"AI triage completed after {elapsed:.1f} seconds")
              break
  
          # Wait before next poll, with exponential backoff
          await asyncio.sleep(poll_interval)
          poll_interval = min(poll_interval * 1.2, max_poll_interval)
  
      # Return the completed triage
      return {
          "success": True,
          "summary": accumulated_response,
          "stream_id": stream_id,
          "metadata": {
              "alert_id": alert_id,
              "output_length": output_length,
              "generation_time_seconds": round(elapsed, 1),
              "prompt_included": prompt is not None,
          },
      }
  
  except Exception as e:
      logger.error(f"Failed to start AI alert triage: {str(e)}")
      return {
          "success": False,
          "message": f"Failed to start AI alert triage: {str(e)}",
      }