mcp-flowise

Instructions

Implementation Reference

• mcp_flowise/server_fastmcp.py:69-98 (handler)

  The primary handler and registration for the 'create_prediction' tool using FastMCP @mcp.tool() decorator. It resolves the chatflow_id from parameter or environment, calls the helper flowise_predict, and returns the raw JSON response or error.
  @mcp.tool() def create_prediction(*, chatflow_id: str = None, question: str) -> str: """ Create a prediction by sending a question to a specific chatflow or assistant. Args: chatflow_id (str, optional): The ID of the chatflow to use. Defaults to FLOWISE_CHATFLOW_ID. question (str): The question or prompt to send to the chatflow. Returns: str: The raw JSON response from Flowise API or an error message if something goes wrong. """ logger.debug(f"create_prediction called with chatflow_id={chatflow_id}, question={question}") chatflow_id = chatflow_id or FLOWISE_CHATFLOW_ID if not chatflow_id and not FLOWISE_ASSISTANT_ID: logger.error("No chatflow_id or assistant_id provided or pre-configured.") return json.dumps({"error": "chatflow_id or assistant_id is required"}) try: # Determine which chatflow ID to use target_chatflow_id = chatflow_id or FLOWISE_ASSISTANT_ID # Call the prediction function and return the raw JSON result result = flowise_predict(target_chatflow_id, question) logger.debug(f"Prediction result: {result}") return result # Returning raw JSON as a string except Exception as e: logger.error(f"Unhandled exception in create_prediction: {e}", exc_info=True) return json.dumps({"error": str(e)})
• mcp_flowise/utils.py:186-227 (helper)

  Core utility function that implements the HTTP request to Flowise API's prediction endpoint, used by the create_prediction handler.
  def flowise_predict(chatflow_id: str, question: str) -> str: """ Sends a question to a specific chatflow ID via the Flowise API and returns the response JSON text. Args: chatflow_id (str): The ID of the Flowise chatflow to be used. question (str): The question or prompt to send to the chatflow. Returns: str: The raw JSON response text from the Flowise API, or an error message if something goes wrong. """ logger = logging.getLogger(__name__) # Construct the Flowise API URL for predictions url = f"{FLOWISE_API_ENDPOINT.rstrip('/')}/api/v1/prediction/{chatflow_id}" headers = { "Content-Type": "application/json", } if FLOWISE_API_KEY: headers["Authorization"] = f"Bearer {FLOWISE_API_KEY}" payload = {"question": question} logger.debug(f"Sending prediction request to {url} with payload: {payload}") try: # Send POST request to the Flowise API response = requests.post(url, json=payload, headers=headers, timeout=30) logger.debug(f"Prediction response code: HTTP {response.status_code}") # response.raise_for_status() # Log the raw response text for debugging logger.debug(f"Raw prediction response: {response.text}") # Return the raw JSON response text return response.text #except requests.exceptions.RequestException as e: except Exception as e: # Log and return an error message logger.error(f"Error during prediction: {e}") return json.dumps({"error": str(e)})
• mcp_flowise/server_lowlevel.py:162-170 (schema)

  Example schema definition for dynamically registered tools in the low-level server (similar to create_prediction inputs). Note: create_prediction specifically uses fastmcp with type hints.
  tool = types.Tool( name=normalized_name, description=description, inputSchema={ "type": "object", "required": ["question"], "properties": {"question": {"type": "string"}}, }, )
• mcp_flowise/server_fastmcp.py:39-39 (registration)

  The @mcp.tool() decorator registers the tool with FastMCP server (same for create_prediction).
  @mcp.tool()
• mcp_flowise/utils.py:229-229 (helper)

  Utility to fetch chatflows, used in related tools but indirectly supports tool ecosystem.
  def fetch_chatflows() -> list[dict]: