get_metabase_current_user

Instructions

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)