get_metabase_current_user

Retrieve details of the currently authenticated Metabase user, including ID, email, and group memberships, for identity verification and access control.

Instructions

Get current logged-in user info from Metabase.

Returns: Dict[str, Any]: User details like id, email, groups, etc.

Input Schema

TableJSON Schema

Name	Required	Description	Default
No arguments

Implementation Reference

src/metabase_mcp_server.py:1250-1259 (handler)
The handler function for 'get_metabase_current_user' tool. This async function retrieves the current logged-in user info from Metabase by making a GET request to the '/api/user/current' endpoint. It takes no parameters and returns a Dict containing user details like id, email, groups, etc. The @mcp.tool() decorator registers it as an MCP tool.
```
@mcp.tool()
async def get_metabase_current_user() -> Dict[str, Any]:
    """
    Get current logged-in user info from Metabase.

    Returns:
        Dict[str, Any]: User details like id, email, groups, etc.
    """
    logger.info("Getting current user info")
    return await make_metabase_request(RequestMethod.GET, "/api/user/current")
```

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

The 'make_metabase_request' helper function that supports the tool handler. This async function makes HTTP requests to the Metabase API using aiohttp. It handles request construction, error handling, response parsing, and ensures responses are properly formatted as dictionaries for FastMCP compatibility. Used by the handler to make the actual GET request to '/api/user/current'.

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/metabase_mcp_server.py:160-160 (registration)
The FastMCP instance initialization that enables tool registration. The 'mcp' object is created with the name 'metabase' and a lifespan manager, and is used to register tools via the @mcp.tool() decorator.
```
mcp = FastMCP("metabase", lifespan=app_lifespan)
```

Metabase MCP Server