adx-mcp-server

Instructions

Implementation Reference

• src/adx_mcp_server/server.py:257-275 (handler)

  The get_table_details tool handler, registered via @mcp.tool decorator. Executes the KQL command '.show table {table_name} details' to fetch table metadata such as TotalRowCount and HotExtentSize, formats the results, and handles errors.
  @mcp.tool(description="Retrieves table details including TotalRowCount, HotExtentSize") async def get_table_details(table_name: str) -> List[Dict[str, Any]]: """Get detailed statistics and metadata for a table.""" logger.info("Getting table details", table_name=table_name, 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".show table {table_name} details" result_set = client.execute(config.database, query) results = format_query_results(result_set) logger.info("Table details retrieved successfully", table_name=table_name) return results except Exception as e: logger.error("Failed to get table details", table_name=table_name, error=str(e), exception_type=type(e).__name__) raise
• src/adx_mcp_server/server.py:138-172 (helper)

  Shared helper function used by get_table_details (and other tools) to format raw Kusto query results into a standardized list of dictionaries.
  def format_query_results(result_set) -> List[Dict[str, Any]]: """ Format Kusto query results into a list of dictionaries. Args: result_set: Raw result set from KustoClient Returns: List of dictionaries with column names as keys """ if not result_set or not result_set.primary_results: logger.debug("Empty or null result set received") return [] try: primary_result = result_set.primary_results[0] columns = [col.column_name for col in primary_result.columns] formatted_results = [] for row in primary_result.rows: record = {} for i, value in enumerate(row): record[columns[i]] = value formatted_results.append(record) logger.debug("Query results formatted", row_count=len(formatted_results), columns=columns) return formatted_results except Exception as e: logger.error( "Error formatting query results", error=str(e), exception_type=type(e).__name__ ) raise
• src/adx_mcp_server/server.py:83-137 (helper)

  Shared helper function used by get_table_details (and other tools) to create and configure the Kusto client with Azure credentials.
  def get_kusto_client() -> KustoClient: """ Create and configure a Kusto client with appropriate Azure credentials. Prioritizes WorkloadIdentityCredential when running in AKS with workload identity, falls back to DefaultAzureCredential for other authentication methods. Returns: KustoClient: Configured Kusto client instance """ tenant_id = os.environ.get('AZURE_TENANT_ID') client_id = os.environ.get('AZURE_CLIENT_ID') token_file_path = os.environ.get('ADX_TOKEN_FILE_PATH', '/var/run/secrets/azure/tokens/azure-identity-token') if tenant_id and client_id: logger.info( "Using WorkloadIdentityCredential", client_id=client_id, tenant_id=tenant_id, token_file_path=token_file_path ) try: credential = WorkloadIdentityCredential( tenant_id=tenant_id, client_id=client_id, token_file_path=token_file_path ) except Exception as e: logger.warning( "Failed to initialize WorkloadIdentityCredential, falling back", error=str(e), exception_type=type(e).__name__ ) credential = DefaultAzureCredential() else: logger.info("Using DefaultAzureCredential (missing WorkloadIdentity credentials)") credential = DefaultAzureCredential() try: kcsb = KustoConnectionStringBuilder.with_azure_token_credential( connection_string=config.cluster_url, credential=credential ) client = KustoClient(kcsb) logger.debug("Kusto client initialized successfully", cluster_url=config.cluster_url) return client except Exception as e: logger.error( "Failed to create Kusto client", error=str(e), exception_type=type(e).__name__, cluster_url=config.cluster_url ) raise
• src/adx_mcp_server/server.py:257-257 (registration)

  The @mcp.tool decorator registers the get_table_details function as an MCP tool with its description used for schema/input validation.
  @mcp.tool(description="Retrieves table details including TotalRowCount, HotExtentSize")