mcp-flowise

Instructions

Implementation Reference

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

  The handler function for the 'list_chatflows' tool. It fetches chatflows using fetch_chatflows(), applies additional whitelist/blacklist filtering based on environment variables, logs the process, and returns a JSON string of the filtered 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)

  Helper function that performs the actual API call to fetch chatflows from Flowise, simplifies the response to id and name, applies internal filtering via filter_chatflows, and returns the list used by the handler.
  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/server_fastmcp.py:39-39 (registration)

  The @mcp.tool() decorator registers the list_chatflows function as an MCP tool.
  @mcp.tool()
• mcp_flowise/utils.py:126-185 (helper)

  Supporting filter function used by fetch_chatflows for initial filtering based on different env vars (FLOWISE_WHITELIST_ID etc.). Note: handler applies additional filtering.
  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