mcp-flowise

Instructions

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.

Implementation Reference

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

  The main handler function for the 'create_prediction' tool, registered via @mcp.tool() decorator. It resolves the chatflow_id, calls the flowise_predict helper, and returns the prediction result as a JSON string.

  @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)

  Supporting utility function that performs the actual HTTP request to the Flowise prediction API 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)})