Qlik Sense MCP Server

Instructions

Implementation Reference

• qlik_sense_mcp_server/engine_api.py:1198-1354 (handler)

  Main handler function that implements the core logic for get_app_field_statistics. Creates a hypercube with statistical expressions (Count DISTINCT, Count, Min/Max/Avg/Sum/Median/Mode/Stdev) for the specified field, extracts results, calculates null/completeness percentages, and returns comprehensive field statistics.
  def get_field_statistics(self, app_handle: int, field_name: str) -> Dict[str, Any]: """Get comprehensive statistics for a field.""" debug_log = [] debug_log.append(f"get_field_statistics called with app_handle={app_handle}, field_name={field_name}") try: # Create expressions for statistics stats_expressions = [ f"Count(DISTINCT [{field_name}])", # Unique values f"Count([{field_name}])", # Total count f"Count({{$<[{field_name}]={{'*'}}>}})", # Non-null count f"Min([{field_name}])", # Minimum value f"Max([{field_name}])", # Maximum value f"Avg([{field_name}])", # Average value f"Sum([{field_name}])", # Sum (if numeric) f"Median([{field_name}])", # Median f"Mode([{field_name}])", # Mode (most frequent) f"Stdev([{field_name}])", # Standard deviation ] debug_log.append(f"Created {len(stats_expressions)} expressions: {stats_expressions}") # Create hypercube for statistics calculation hypercube_def = { "qDimensions": [], "qMeasures": [ {"qDef": {"qDef": expr, "qLabel": f"Stat_{i}"}} for i, expr in enumerate(stats_expressions) ], "qInitialDataFetch": [ { "qTop": 0, "qLeft": 0, "qHeight": 1, "qWidth": len(stats_expressions), } ], "qSuppressZero": False, "qSuppressMissing": False, } obj_def = { "qInfo": {"qId": f"field-stats-{field_name}", "qType": "HyperCube"}, "qHyperCubeDef": hypercube_def, } # Create session object debug_log.append(f"Creating session object with obj_def: {obj_def}") result = self.send_request( "CreateSessionObject", [obj_def], handle=app_handle ) debug_log.append(f"CreateSessionObject result: {result}") if "qReturn" not in result or "qHandle" not in result["qReturn"]: debug_log.append(f"Failed to create session object, returning error") return { "error": "Failed to create statistics hypercube", "response": result, "debug_log": debug_log } cube_handle = result["qReturn"]["qHandle"] # Get layout with data layout = self.send_request("GetLayout", [], handle=cube_handle) if "qLayout" not in layout or "qHyperCube" not in layout["qLayout"]: try: self.send_request( "DestroySessionObject", [f"field-stats-{field_name}"], handle=app_handle, ) except: pass return {"error": "No hypercube in statistics layout", "layout": layout, "debug_log": debug_log} hypercube = layout["qLayout"]["qHyperCube"] # Extract statistics values stats_labels = [ "unique_values", "total_count", "non_null_count", "min_value", "max_value", "avg_value", "sum_value", "median_value", "mode_value", "std_deviation", ] statistics = {"field_name": field_name} for page in hypercube.get("qDataPages", []): for row in page.get("qMatrix", []): for i, cell in enumerate(row): if i < len(stats_labels): stat_name = stats_labels[i] statistics[stat_name] = { "text": cell.get("qText", ""), "numeric": ( cell.get("qNum", None) if cell.get("qNum") != "NaN" else None ), "is_numeric": cell.get("qIsNumeric", False), } # Calculate additional derived statistics debug_log.append(f"Statistics before calculation: {statistics}") if "total_count" in statistics and "non_null_count" in statistics: # Handle None values safely total_dict = statistics["total_count"] non_null_dict = statistics["non_null_count"] debug_log.append(f"total_dict: {total_dict}") debug_log.append(f"non_null_dict: {non_null_dict}") total = total_dict.get("numeric", 0) if total_dict.get("numeric") is not None else 0 non_null = non_null_dict.get("numeric", 0) if non_null_dict.get("numeric") is not None else 0 debug_log.append(f"total: {total} (type: {type(total)})") debug_log.append(f"non_null: {non_null} (type: {type(non_null)})") if total > 0: debug_log.append(f"Calculating percentages...") debug_log.append(f"Calculation: ({total} - {non_null}) / {total} * 100") statistics["null_percentage"] = round( (total - non_null) / total * 100, 2 ) statistics["completeness_percentage"] = round( non_null / total * 100, 2 ) debug_log.append(f"Percentages calculated successfully") # Cleanup try: self.send_request( "DestroySessionObject", [f"field-stats-{field_name}"], handle=app_handle, ) except Exception as cleanup_error: statistics["cleanup_warning"] = str(cleanup_error) statistics["debug_log"] = debug_log return statistics except Exception as e: import traceback debug_log.append(f"Exception in get_field_statistics: {e}") debug_log.append(f"Traceback: {traceback.format_exc()}") return { "error": str(e), "details": "Error in get_field_statistics method", "traceback": traceback.format_exc(), "debug_log": debug_log }
• qlik_sense_mcp_server/server.py:592-634 (handler)

  Server-side dispatch handler in call_tool that validates arguments, connects to Qlik Engine, opens the app safely, calls engine_api.get_field_statistics, handles errors with debug info, and returns JSON-formatted TextContent response.
  elif name == "get_app_field_statistics": app_id = arguments["app_id"] field_name = arguments["field_name"] def _get_field_statistics(): app_handle = -1 debug_info = [] try: debug_info.append(f"Starting field statistics for app_id={app_id}, field_name={field_name}") self.engine_api.connect() debug_info.append("Connected to engine") app_result = self.engine_api.open_doc_safe(app_id, no_data=False) debug_info.append(f"App open result: {app_result}") app_handle = app_result.get("qReturn", {}).get("qHandle", -1) debug_info.append(f"App handle: {app_handle}") if app_handle != -1: result = self.engine_api.get_field_statistics(app_handle, field_name) debug_info.append("Field statistics method completed") if isinstance(result, dict) and "debug_log" not in result: result["server_debug"] = debug_info return result else: raise Exception(f"Failed to open app: {app_result}") except Exception as e: import traceback debug_info.append(f"Exception in server handler: {e}") debug_info.append(f"Traceback: {traceback.format_exc()}") return { "error": str(e), "server_debug": debug_info, "traceback": traceback.format_exc() } finally: debug_info.append("Disconnecting from engine") self.engine_api.disconnect() result = await asyncio.to_thread(_get_field_statistics) return [ TextContent( type="text", text=json.dumps(result, indent=2, ensure_ascii=False) ) ]
• qlik_sense_mcp_server/server.py:276-276 (registration)

  MCP tool registration in list_tools() handler, defining the tool name, description, and input schema requiring app_id and field_name.
  Tool(name="get_app_field_statistics", description="Get comprehensive statistics for a field", inputSchema={"type": "object", "properties": {"app_id": {"type": "string", "description": "Application ID"}, "field_name": {"type": "string", "description": "Field name"}}, "required": ["app_id", "field_name"]}),
• qlik_sense_mcp_server/server.py:276-276 (schema)

  Input schema definition for the get_app_field_statistics tool, specifying object with required string properties app_id and field_name.
  Tool(name="get_app_field_statistics", description="Get comprehensive statistics for a field", inputSchema={"type": "object", "properties": {"app_id": {"type": "string", "description": "Application ID"}, "field_name": {"type": "string", "description": "Field name"}}, "required": ["app_id", "field_name"]}),