Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/alerts.py:1118-1213 (handler)

  The core handler function that implements the get_ai_alert_triage_summary tool. It fetches the latest AI-generated triage summary for a given alert ID by querying Panther's GraphQL API for AI inference streams metadata and then retrieving the response from the most recent stream.
  async def get_ai_alert_triage_summary( alert_id: Annotated[ str, Field( min_length=1, description="The ID of the alert to retrieve the latest AI triage summary for", ), ], ) -> dict[str, Any]: """Retrieve the latest AI triage summary for a specific Panther alert. This tool retrieves the most recently generated AI triage analysis for an alert. It fetches the list of AI inference stream IDs associated with the alert, then retrieves the response text for the latest stream. Returns: Dict containing: - success: Boolean indicating if retrieval was successful - summary: The latest AI triage summary containing: - stream_id: The unique stream identifier - response_text: The AI-generated triage summary - finished: Whether the triage generation completed - error: Any error message if present - message: Error message if unsuccessful """ logger.info(f"Retrieving latest AI triage summary for alert {alert_id}") try: # Step 1: Get all stream IDs for this alert metadata_variables = {"input": {"alias": alert_id}} metadata_result = await _execute_query( AI_INFERENCE_STREAMS_METADATA_QUERY, metadata_variables ) if not metadata_result or "aiInferenceStreamsMetadata" not in metadata_result: logger.error("Failed to retrieve AI inference streams metadata") return { "success": False, "message": "Failed to retrieve AI inference streams metadata", } edges = metadata_result["aiInferenceStreamsMetadata"].get("edges", []) stream_ids = [edge["node"]["streamId"] for edge in edges] if not stream_ids: logger.info(f"No AI triage summary found for alert {alert_id}") return { "success": False, "message": "No AI triage summary found for this alert", } # Get the latest stream ID (last in the list) latest_stream_id = stream_ids[-1] logger.info( f"Found {len(stream_ids)} AI triage summary/summaries, retrieving latest: {latest_stream_id}" ) # Step 2: Fetch response text for the latest stream ID stream_variables = {"streamId": latest_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(f"Failed to retrieve stream {latest_stream_id}") return { "success": False, "message": f"Failed to retrieve stream data for {latest_stream_id}", } inference_data = stream_result["aiInferenceStream"] response_text = inference_data.get("responseText", "") error = inference_data.get("error") summary = { "stream_id": latest_stream_id, "response_text": response_text, "finished": inference_data.get("finished", False), "error": error, } logger.info( f"Successfully retrieved latest AI triage summary for alert {alert_id}" ) return { "success": True, "summary": summary, } except Exception as e: logger.error(f"Failed to retrieve AI alert triage summaries: {str(e)}") return { "success": False, "message": f"Failed to retrieve AI alert triage summaries: {str(e)}", }
• src/mcp_panther/panther_mcp_core/tools/alerts.py:1112-1117 (registration)

  The @mcp_tool decorator that registers the get_ai_alert_triage_summary function as an MCP tool, specifying required permissions and read-only nature.
  @mcp_tool( annotations={ "permissions": all_perms(Permission.RUN_PANTHER_AI), "readOnlyHint": True, } )
• src/mcp_panther/panther_mcp_core/tools/alerts.py:1119-1126 (schema)

  Pydantic schema definition for the tool's single input parameter: alert_id (string, required).
  alert_id: Annotated[ str, Field( min_length=1, description="The ID of the alert to retrieve the latest AI triage summary for", ), ], ) -> dict[str, Any]: