get_table_details

Retrieve table details like row count and extent size from Azure Data Explorer databases to analyze data structure and performance.

Instructions

Retrieves table details including TotalRowCount, HotExtentSize

Input Schema

TableJSON Schema

Name	Required	Description	Default
`table_name`	Yes

Output Schema

TableJSON Schema

Name	Required	Description	Default
`result`	Yes

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")
```

adx-mcp-server