get_card_query_results

Retrieve query results from a Metabase card by providing its ID to access business intelligence data through the MCP server.

Instructions

Get the results of a card's query.

Args: card_id (int): ID of the card.

Returns: Dict[str, Any]: Query result data.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`card_id`	Yes

Implementation Reference

src/metabase_mcp_server.py:335-347 (handler)

The main handler function for 'get_card_query_results' tool. It's decorated with @mcp.tool() to register it as an MCP tool. The function takes a card_id (int) and makes a POST request to the Metabase API endpoint /api/card/{card_id}/query to retrieve query results.

@mcp.tool()
async def get_card_query_results(card_id: int) -> Dict[str, Any]:
    """
    Get the results of a card's query.
    
    Args:
        card_id (int): ID of the card.
        
    Returns:
        Dict[str, Any]: Query result data.
    """
    logger.info(f"Getting query results for card {card_id}")
    return await make_metabase_request(RequestMethod.POST, f"/api/card/{card_id}/query")

src/metabase_mcp_server.py:159-160 (registration)
Initialization of the FastMCP instance named 'metabase' which serves as the MCP server. Tools are registered via the @mcp.tool() decorator.
```
# Initialize FastMCP agent
mcp = FastMCP("metabase", lifespan=app_lifespan)
```

src/metabase_mcp_server.py:162-247 (helper)

The 'make_metabase_request' helper function that handles all HTTP communication with the Metabase API. It manages session handling, authentication via API key, error handling with custom exceptions, and response parsing.

async def make_metabase_request(
    method: RequestMethod,
    endpoint: str,
    data: Optional[Dict[str, Any] | bytes] = None,
    params: Optional[Dict[str, Any]] = None,
    json: Any = None,
    headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    """
    Make a request to the Metabase API.
    
    Args:
        method: HTTP method to use (GET, POST, PUT, DELETE)
        endpoint: API endpoint path
        data: Request data (for form data)
        params: URL parameters
        json: JSON request body
        headers: Additional headers
        
    Returns:
        Dict[str, Any]: Response data
        
    Raises:
        MetabaseConnectionError: When the Metabase server is unreachable
        MetabaseResponseError: When Metabase returns a non-2xx status code
        RuntimeError: For other errors
    """
    
    if not METABASE_URL or not METABASE_API_KEY:
        raise RuntimeError("METABASE_URL or METABASE_API_KEY environment variable is not set. Metabase API requests will fail.")

    if session is None:
        raise RuntimeError("HTTP session is not initialized. Ensure app_lifespan was called.")

    try:
        request_headers = headers or {}
        
        logger.debug(f"Making {method.name} request to {METABASE_URL}{endpoint}")
        
        # Log request payload for debugging (omit sensitive info)
        if json and logger.level <= logging.DEBUG:
            sanitized_json = {**json}
            if 'password' in sanitized_json:
                sanitized_json['password'] = '********'
            logger.debug(f"Request payload: {sanitized_json}")
            
        response = await session.request(
            method=method.name,
            url=endpoint,
            timeout=aiohttp.ClientTimeout(total=30),
            headers=request_headers,
            data=data,
            params=params,
            json=json,
        )

        try:
            # Handle 500 errors with more detailed info
            if response.status >= 500:
                error_text = await response.text()
                logger.error(f"Server error {response.status}: {error_text[:200]}")
                raise MetabaseResponseError(response.status, f"Server Error: {error_text[:200]}", endpoint)
            
            response.raise_for_status()
            response_data = await response.json()
            
            # Ensure the response is a dictionary for FastMCP compatibility
            return ensure_dict_response(response_data)
            
        except aiohttp.ContentTypeError:
            # Handle empty responses or non-JSON responses
            content = await response.text()
            if not content:
                return {"data": {}}
            logger.warning(f"Received non-JSON response: {content}")
            return {"data": content}

    except aiohttp.ClientConnectionError as e:
        logger.error(f"Connection error: {str(e)}")
        raise MetabaseConnectionError("Metabase is unreachable. Is the Metabase server running?") from e
    except aiohttp.ClientResponseError as e:
        logger.error(f"Response error: {e.status}, {e.message}, {e.request_info.url}")
        raise MetabaseResponseError(e.status, e.message, str(e.request_info.url)) from e
    except Exception as e:
        logger.error(f"Request error: {str(e)}")
        raise RuntimeError(f"Request error: {str(e)}") from e

src/enums/request_enum.py:1-11 (helper)
RequestMethod enum defining HTTP methods (GET, POST, PUT, DELETE) used by the tool to specify the request type when calling make_metabase_request.
```
from enum import Enum, auto

class RequestMethod(Enum):
    GET = auto()
    POST = auto()
    PUT = auto()
    DELETE = auto()

    def __str__(self):
        return self.name
```

Metabase MCP Server