adx-mcp-server

Instructions

Implementation Reference

• src/adx_mcp_server/server.py:217-235 (handler)

  The main handler function for the 'get_table_schema' tool. It is registered via the @mcp.tool decorator and executes a KQL query to retrieve and format the table schema.
  @mcp.tool(description="Retrieves the schema information for a specified table in the Azure Data Explorer database, including column names, data types, and other schema-related metadata.") async def get_table_schema(table_name: str) -> List[Dict[str, Any]]: """Get schema information for a specific table.""" logger.info("Getting table schema", 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"{table_name} | getschema" result_set = client.execute(config.database, query) results = format_query_results(result_set) logger.info("Schema retrieved successfully", table_name=table_name, column_count=len(results)) return results except Exception as e: logger.error("Failed to get table schema", table_name=table_name, error=str(e), exception_type=type(e).__name__) raise
• src/adx_mcp_server/server.py:138-171 (helper)

  Helper function used by get_table_schema to format the raw Kusto query results into a 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-136 (helper)

  Helper function used by get_table_schema to create and configure the Kusto client for querying ADX.
  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