Import json file as custom component with python

[Tuguldur-Tserenbaljir]

I want to ask is there a way to Import json file as custom component with python.

def upload_to_langflow(file_path: str, langflow_url: str = ""):
    #Removed the langflowurl for privacy 
    
    upload_url = f"{langflow_url}/api/v1/flows/"
    with open(file_path, 'r') as f:
        files = {'file': (file_path, f.read(), 'application/json')}
    response = requests.post(upload_url, files=files)
    response.raise_for_status()
    return response.json()

if __name__ == "__main__":
    result = upload_to_langflow("component.json")
    print("Upload successful!", result)

So far I have been to come up with this kind of theory but the response is:

Traceback (most recent call last):
  File "c:\Users\btuud\Downloads\visual_studio_tests\test.py", line 13, in <module>
    result = upload_to_langflow("component.json")
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "c:\Users\btuud\Downloads\visual_studio_tests\test.py", line 9, in upload_to_langflow
    response.raise_for_status()
  File "C:\Users\btuud\anaconda3\Lib\site-packages\requests\models.py", line 1021, in raise_for_status
    raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 422 Client Error: Unprocessable Entity for URL:

[dosubot[bot]]

A 422 Unprocessable Entity error typically indicates that the data being sent to the API is not in the expected format or structure. In the context of Langflow's API, common causes for this error include:

1. Invalid JSON Data: Ensure that the JSON file you are uploading contains valid JSON data with the required structure, including "nodes" and "edges" keys ^[1].

2. Invalid Input Format: The input format might be incorrect. Make sure that the JSON data adheres to the expected schema for Langflow components ^[1].

3. Authentication Requirements: Ensure that you are including the necessary authentication headers, such as an "Authorization" header with a Bearer token, or using API key authentication ^[2].

To resolve the issue, verify that your JSON file matches the expected format and structure for Langflow components. Additionally, ensure that you are providing the correct authentication details in your request.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[viniciossilva3]

Hi @Tuguldur-Tserenbaljir , try using the example below!
If this helps you, mark the answer as correct so it can help other people too.

import requests
import json
from datetime import datetime

def upload_flow(json_file_path, langflow_api_url="http://localhost:7860/api/v1/flows/"):
    """
    Upload a JSON flow to Langflow API
    
    Args:
        json_file_path (str): Path to the JSON flow file
        langflow_api_url (str): Langflow API URL (default: http://localhost:7860/api/v1/flows/)
        
    Returns:
        dict: Response from the API
    """
    print("\n" + "="*50)
    print(f"🚀 Starting flow upload at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="*50)
    
    try:
        # Read the JSON flow file
        print(f"\n📂 Reading flow from: {json_file_path}")
        with open(json_file_path, 'r') as file:
            flow_data = json.load(file)
        
        # Make the POST request to Langflow API
        print(f"📡 Sending request to: {langflow_api_url}")
        response = requests.post(
            langflow_api_url,
            json=flow_data,
            headers={'Content-Type': 'application/json'}
        )
        
        # Check if the request was successful
        response.raise_for_status()
        
        return response.json()
    
    except FileNotFoundError:
        print(f"\n❌ Error: The file {json_file_path} was not found.")
        return None
    except json.JSONDecodeError:
        print(f"\n❌ Error: The file {json_file_path} is not a valid JSON file.")
        return None
    except requests.exceptions.RequestException as e:
        print(f"\n❌ Error making the request to Langflow API: {str(e)}")
        return None

if __name__ == "__main__":
    # Example usage
    json_file_path = "path/to/your/flow.json"
    result = upload_flow(json_file_path)
    
    if result:
        print("\n✅ Flow uploaded successfully!")
        print("\n📊 Response Details:")
        print("-"*50)
        for key, value in result.items():
            print(f"{key}: {value}")
        print("-"*50)
    else:
        print("\n❌ Failed to upload flow. Please check the errors above.")

[Tuguldur-Tserenbaljir]

Thank you, it works and answers my question!

I have a follow-up: Is it possible to directly import this component into an existing flow?

For instance, if my flow's URL is http://localhost:7860/flow/4f362e03-d4db-40be-8ef3-8ee8c1564480, can I upload the component directly into the flow with the ID 4f362e03-d4db-40be-8ef3-8ee8c1564480?

[viniciossilva3]

Hi @Tuguldur-Tserenbaljir! Yes, it is possible, you can test it as in the example below!
The code is functional, but be very careful with the component structure settings, if you modify them, problems may occur such as the component appearing empty or other problems.

output.mp4

import requests
import json
from datetime import datetime
from typing import Optional, Dict, Any
from uuid import uuid4


def extract_component_info(component_data: dict) -> tuple[Optional[dict], Optional[str], Optional[dict]]:
    """
    Extract component information from the JSON data
    
    Args:
        component_data (dict): The component JSON data
        
    Returns:
        tuple: (component_node, component_type, component_data)
    """
    try:
        # Get the first node from the component data
        nodes = component_data.get("data", {}).get("nodes", [])
        if not nodes:
            return None, None, None
            
        node = nodes[0]
        node_data = node.get("data", {})
        
        # Extract component info
        component_type = node_data.get("type", "")
        if not component_type:
            return None, None, None
            
        return node_data.get("node", {}), component_type, node
        
    except Exception as e:
        print(f"Error extracting component info: {str(e)}")
        return None, None, None


def add_component_to_flow(
    component_json_path: str,
    flow_id: str,
    position: Dict[str, int] = {"x": 100, "y": 100},
    langflow_api_url: str = "http://localhost:7860/api/v1/flows/"
) -> Optional[dict]:
    """
    Add a component to an existing flow in Langflow
    
    Args:
        component_json_path (str): Path to the component JSON file
        flow_id (str): ID of the existing flow
        position (dict): Position where to place the component (x, y coordinates)
        langflow_api_url (str): Base Langflow API URL
        
    Returns:
        dict: Response from the API
    """
    print("\n" + "="*50)
    print(f"🚀 Adding component to flow at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("="*50)
    
    try:
        # First, get the existing flow
        flow_endpoint = f"{langflow_api_url.rstrip('/')}/{flow_id}"
        print(f"\n📡 Fetching existing flow from: {flow_endpoint}")
        
        response = requests.get(flow_endpoint)
        response.raise_for_status()
        flow_data = response.json()
        
        # Read the component JSON
        print(f"\n📂 Reading component from: {component_json_path}")
        with open(component_json_path, 'r') as file:
            component_data = json.load(file)
        
        # Extract component info
        component_node, component_type, node_template = extract_component_info(component_data)
        if not component_node or not component_type or not node_template:
            print("\n❌ Error: Could not extract component information from JSON")
            return None
            
        print(f"\n🔍 Found component type: {component_type}")
            
        # Create node in the format expected by Langflow
        new_id = f"{component_type}-{str(uuid4())[:6]}"
        
        # Start with the template from the component
        new_node = {
            "id": new_id,
            "type": "genericNode",
            "position": position,
            "data": {
                "node": component_node,
                "id": new_id,
                "type": component_type
            }
        }
        
        # Copy additional fields from the template
        for field in ["selected", "width", "height", "dragging", "positionAbsolute"]:
            if field in node_template:
                new_node[field] = node_template[field]
                
        # Copy additional data fields from the template
        for field in ["value", "showNode", "display_name", "description"]:
            if field in node_template.get("data", {}):
                new_node["data"][field] = node_template["data"][field]
            
        print(f"\n📋 Component structure:")
        print(json.dumps(new_node, indent=2))
        
        # Add the component to the flow's data
        if "data" in flow_data and "nodes" in flow_data["data"]:
            flow_data["data"]["nodes"].append(new_node)
        else:
            print("\n❌ Error: Invalid flow data structure")
            return None
        
        # Update the flow with the new component
        print(f"\n📡 Updating flow with new component")
        update_endpoint = f"{langflow_api_url.rstrip('/')}/{flow_id}"
        update_response = requests.patch(
            update_endpoint,
            json=flow_data,
            headers={'Content-Type': 'application/json'}
        )
        
        update_response.raise_for_status()
        return update_response.json()
    
    except FileNotFoundError:
        print(f"\n❌ Error: The component file {component_json_path} was not found.")
        return None
    except json.JSONDecodeError:
        print(f"\n❌ Error: The file {component_json_path} is not a valid JSON file.")
        return None
    except requests.exceptions.RequestException as e:
        print(f"\n❌ Error making the request to Langflow API: {str(e)}")
        if hasattr(e.response, 'text'):
            print(f"Response details: {e.response.text}")
        return None


if __name__ == "__main__":
    # Example usage
    component_path = "/path/to/component.json"
    flow_id = "your-flow-id"
    
    # Optionally specify position (default is x:100, y:100)
    custom_position = {
        "x": 200,
        "y": 150
    }
    
    result = add_component_to_flow(
        component_json_path=component_path,
        flow_id=flow_id,
        position=custom_position
    )
    
    if result:
        print("\n✅ Component successfully added to flow!")
        print("\n📊 Response Details:")
        print("-"*50)
        for key, value in result.items():
            print(f"{key}: {value}")
        print("-"*50)
    else:
        print("\n❌ Failed to add component to flow. Please check the errors above.")

[Tuguldur-Tserenbaljir]

Thank you very much!