Metabase MCP Server

Instructions

Implementation Reference

• src/metabase_mcp_server.py:1147-1186 (handler)

  The create_metabase_user tool handler function. Decorated with @mcp.tool(), it accepts user details (first_name, last_name, email, password) and optional parameters (login_attributes, group_ids, is_superuser), constructs a payload, and makes a POST request to /api/user endpoint to create a new Metabase user.

  @mcp.tool()
  async def create_metabase_user(
      first_name: str,
      last_name: str,
      email: str,
      password: str,
      login_attributes: Optional[Dict[str, Any]] = None,
      group_ids: Optional[List] = None,
      is_superuser: Optional[bool] = None
  ) -> Dict[str, Any]:
      """
      Create a new user in Metabase.
  
      Args:
          first_name (str): User's first name.
          last_name (str): User's last name.
          email (str): Email address.
          password (str): Account password.
          login_attributes (dict, optional): Additional login metadata.
          group_ids (list, optional): List of group IDs to assign the user.
          is_superuser (bool, optional): Whether the user is a superuser.
  
      Returns:
          Dict[str, Any]: Created user metadata.
      """
      payload = {
          "first_name": first_name,
          "last_name": last_name,
          "email": email,
          "password": password,
      }
      if login_attributes is not None:
          payload["login_attributes"] = login_attributes
      if group_ids is not None:
          payload["group_ids"] = group_ids
      if is_superuser is not None:
          payload["is_superuser"] = is_superuser
  
      logger.info(f"Creating user '{email}'")
      return await make_metabase_request(RequestMethod.POST, "/api/user", json=payload)

• src/metabase_mcp_server.py:1147-1147 (registration)

  The @mcp.tool() decorator registers the create_metabase_user function as an MCP tool, making it available to clients via the Model Context Protocol.

  @mcp.tool()

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

  The make_metabase_request helper function that handles all HTTP communication with the Metabase API. It validates environment variables, manages the HTTP session, handles request/response processing, error handling, and sanitizes sensitive data (like passwords) in logs. This function is called by create_metabase_user to make the actual API request.

  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:1148-1156 (schema)

  Type annotations defining the input schema for create_metabase_user. Parameters include required fields (first_name: str, last_name: str, email: str, password: str) and optional fields (login_attributes: Optional[Dict[str, Any]], group_ids: Optional[List], is_superuser: Optional[bool]). FastMCP uses these type hints for automatic schema generation and validation.

  async def create_metabase_user(
      first_name: str,
      last_name: str,
      email: str,
      password: str,
      login_attributes: Optional[Dict[str, Any]] = None,
      group_ids: Optional[List] = None,
      is_superuser: Optional[bool] = None
  ) -> Dict[str, Any]: