mcp-flowise

Instructions

List all available chatflows from the Flowise API.

This function respects optional whitelisting or blacklisting if configured
via FLOWISE_CHATFLOW_WHITELIST or FLOWISE_CHATFLOW_BLACKLIST.

Returns:
    str: A JSON-encoded string of filtered chatflows.

Implementation Reference

• mcp_flowise/server_fastmcp.py:39-66 (handler)

  The primary handler and registration for the 'list_chatflows' tool using FastMCP's @mcp.tool() decorator. It fetches chatflows, applies custom whitelist/blacklist filtering, and returns a JSON string of available chatflows.

  @mcp.tool()
  def list_chatflows() -> str:
      """
      List all available chatflows from the Flowise API.
  
      This function respects optional whitelisting or blacklisting if configured
      via FLOWISE_CHATFLOW_WHITELIST or FLOWISE_CHATFLOW_BLACKLIST.
  
      Returns:
          str: A JSON-encoded string of filtered chatflows.
      """
      logger.debug("Handling list_chatflows tool.")
      chatflows = fetch_chatflows()
  
      # Apply whitelisting
      if FLOWISE_CHATFLOW_WHITELIST:
          whitelist = set(FLOWISE_CHATFLOW_WHITELIST.split(","))
          chatflows = [cf for cf in chatflows if cf["id"] in whitelist]
          logger.debug(f"Applied whitelist filter: {whitelist}")
  
      # Apply blacklisting
      if FLOWISE_CHATFLOW_BLACKLIST:
          blacklist = set(FLOWISE_CHATFLOW_BLACKLIST.split(","))
          chatflows = [cf for cf in chatflows if cf["id"] not in blacklist]
          logger.debug(f"Applied blacklist filter: {blacklist}")
  
      logger.debug(f"Filtered chatflows: {chatflows}")
      return json.dumps(chatflows)

• mcp_flowise/utils.py:229-263 (helper)

  Core helper function called by the list_chatflows handler to retrieve and filter chatflows from the Flowise API.

  def fetch_chatflows() -> list[dict]:
      """
      Fetch a list of all chatflows from the Flowise API.
  
      Returns:
          list of dict: Each dict contains the 'id' and 'name' of a chatflow.
                        Returns an empty list if there's an error.
      """
      logger = logging.getLogger(__name__)
  
      # Construct the Flowise API URL for fetching chatflows
      url = f"{FLOWISE_API_ENDPOINT.rstrip('/')}/api/v1/chatflows"
      headers = {}
      if FLOWISE_API_KEY:
          headers["Authorization"] = f"Bearer {FLOWISE_API_KEY}"
  
      logger.debug(f"Fetching chatflows from {url}")
  
      try:
          # Send GET request to the Flowise API
          response = requests.get(url, headers=headers, timeout=30)
          response.raise_for_status()
  
          # Parse and simplify the response data
          chatflows_data = response.json()
          simplified_chatflows = [{"id": cf["id"], "name": cf["name"]} for cf in chatflows_data]
  
          logger.debug(f"Fetched chatflows: {simplified_chatflows}")
          return filter_chatflows(simplified_chatflows)
      #except requests.exceptions.RequestException as e:
      except Exception as e:
          # Log and return an empty list on error
          logger.error(f"Error fetching chatflows: {e}")
          return []

• mcp_flowise/utils.py:126-185 (helper)

  Supporting filter function used by fetch_chatflows to apply whitelist and blacklist rules based on environment variables.

  def filter_chatflows(chatflows: list[dict]) -> list[dict]:
      """
      Filters chatflows based on whitelist and blacklist criteria.
      Whitelist takes precedence over blacklist.
  
      Args:
          chatflows (list[dict]): A list of chatflow dictionaries.
  
      Returns:
          list[dict]: Filtered list of chatflows.
      """
      logger = logging.getLogger(__name__)
  
      # Dynamically fetch filtering criteria
      whitelist_ids = set(filter(bool, os.getenv("FLOWISE_WHITELIST_ID", "").split(",")))
      blacklist_ids = set(filter(bool, os.getenv("FLOWISE_BLACKLIST_ID", "").split(",")))
      whitelist_name_regex = os.getenv("FLOWISE_WHITELIST_NAME_REGEX", "")
      blacklist_name_regex = os.getenv("FLOWISE_BLACKLIST_NAME_REGEX", "")
  
      filtered_chatflows = []
  
      for chatflow in chatflows:
          chatflow_id = chatflow.get("id", "")
          chatflow_name = chatflow.get("name", "")
  
          # Flags to determine inclusion
          is_whitelisted = False
  
          # Check Whitelist
          if whitelist_ids or whitelist_name_regex:
              if whitelist_ids and chatflow_id in whitelist_ids:
                  is_whitelisted = True
              if whitelist_name_regex and re.search(whitelist_name_regex, chatflow_name):
                  is_whitelisted = True
  
              if is_whitelisted:
                  # If whitelisted, include regardless of blacklist
                  logger.debug("Including whitelisted chatflow '%s' (ID: '%s').", chatflow_name, chatflow_id)
                  filtered_chatflows.append(chatflow)
                  continue  # Skip blacklist checks
              else:
                  # If not whitelisted, exclude regardless of blacklist
                  logger.debug("Excluding non-whitelisted chatflow '%s' (ID: '%s').", chatflow_name, chatflow_id)
                  continue
          else:
              # If no whitelist, apply blacklist directly
              if blacklist_ids and chatflow_id in blacklist_ids:
                  logger.debug("Skipping chatflow '%s' (ID: '%s') - In blacklist.", chatflow_name, chatflow_id)
                  continue  # Exclude blacklisted by ID
              if blacklist_name_regex and re.search(blacklist_name_regex, chatflow_name):
                  logger.debug("Skipping chatflow '%s' (ID: '%s') - Name matches blacklist regex.", chatflow_name, chatflow_id)
                  continue  # Exclude blacklisted by name
  
              # Include the chatflow if it passes all filters
              logger.debug("Including chatflow '%s' (ID: '%s').", chatflow_name, chatflow_id)
              filtered_chatflows.append(chatflow)
  
      logger.debug("Filtered chatflows: %d out of %d", len(filtered_chatflows), len(chatflows))
      return filtered_chatflows