adx-mcp-server

Instructions

Implementation Reference

• src/adx_mcp_server/server.py:237-255 (handler)

  The handler function for the 'sample_table_data' tool, which is automatically registered via the @mcp.tool decorator. It samples rows from a specified ADX table using a KQL query and formats the results into a list of dictionaries.

  @mcp.tool(description="Retrieves a random sample of rows from the specified table in the Azure Data Explorer database. The sample_size parameter controls how many rows to return (default: 10).")
  async def sample_table_data(table_name: str, sample_size: int = 10) -> List[Dict[str, Any]]:
      """Get sample data from a table."""
      logger.info("Sampling table data", table_name=table_name, sample_size=sample_size, database=config.database)
  
      if not config.cluster_url or not config.database:
          logger.error("Missing ADX configuration")
          raise ValueError("Azure Data Explorer configuration is missing. Please set ADX_CLUSTER_URL and ADX_DATABASE environment variables.")
  
      try:
          client = get_kusto_client()
          query = f"{table_name} | sample {sample_size}"
          result_set = client.execute(config.database, query)
          results = format_query_results(result_set)
          logger.info("Sample data retrieved successfully", table_name=table_name, row_count=len(results))
          return results
      except Exception as e:
          logger.error("Failed to sample table data", table_name=table_name, error=str(e), exception_type=type(e).__name__)
          raise