feat: make it possible to load graphs using get_graph function in scripts

src/backend/tests/data/simple_agent.py

@@ -19,29 +19,41 @@
 
 from lfx.graph import Graph
 from lfx.log.logger import LogConfig
-from lfx.utils.async_helpers import run_until_complete
 
 # Using the new flattened component access
 from lfx import components as cp
 
-log_config = LogConfig(
-    log_level="INFO",
-    log_file=Path("langflow.log"),
-)
-
-# Showcase the new flattened component access - no need for deep imports!
-chat_input = cp.ChatInput()
-agent = cp.AgentComponent()
-url_component = cp.URLComponent()
-tools = run_until_complete(url_component.to_toolkit())
-
-agent.set(
-    model_name="gpt-4o-mini",
-    agent_llm="OpenAI",
-    api_key=os.getenv("OPENAI_API_KEY"),
-    input_value=chat_input.message_response,
-    tools=tools,
-)
-chat_output = cp.ChatOutput().set(input_value=agent.message_response)
-
-graph = Graph(chat_input, chat_output, log_config=log_config)
+
+async def get_graph() -> Graph:
+    """Create and return the graph with async component initialization.
+
+    This function properly handles async component initialization without
+    blocking the module loading process. The script loader will detect this
+    async function and handle it appropriately using run_until_complete.
+
+    Returns:
+        Graph: The configured graph with ChatInput → Agent → ChatOutput flow
+    """
+    log_config = LogConfig(
+        log_level="INFO",
+        log_file=Path("langflow.log"),
+    )
+
+    # Showcase the new flattened component access - no need for deep imports!
+    chat_input = cp.ChatInput()
+    agent = cp.AgentComponent()
+
+    # Use URLComponent for web search capabilities
+    url_component = cp.URLComponent()
+    tools = await url_component.to_toolkit()
+
+    agent.set(
+        model_name="gpt-4o-mini",
+        agent_llm="OpenAI",
+        api_key=os.getenv("OPENAI_API_KEY"),
+        input_value=chat_input.message_response,
+        tools=tools,
+    )
+    chat_output = cp.ChatOutput().set(input_value=agent.message_response)
+
+    return Graph(chat_input, chat_output, log_config=log_config)

src/backend/tests/unit/test_simple_agent_in_lfx_run.py

@@ -25,14 +25,15 @@ def simple_agent_script_content(self):
         return '''"""A simple agent flow example for Langflow.
 
 This script demonstrates how to set up a conversational agent using Langflow's
-Agent component with web search capabilities.
+Agent component with proper async handling.
 
 Features:
 - Uses the new flattened component access (cp.AgentComponent instead of deep imports)
 - Configures logging to 'langflow.log' at INFO level
 - Creates an agent with OpenAI GPT model
-- Provides web search tools via URLComponent
 - Connects ChatInput → Agent → ChatOutput
+- Uses async get_graph() function for proper async handling
+- Demonstrates the new async script loading pattern
 
 Usage:
     uv run lfx run simple_agent.py "How are you?"
@@ -46,27 +47,42 @@ def simple_agent_script_content(self):
 from lfx.graph import Graph
 from lfx.log.logger import LogConfig
 
-log_config = LogConfig(
-    log_level="INFO",
-    log_file=Path("langflow.log"),
-)
-
-# Showcase the new flattened component access - no need for deep imports!
-chat_input = cp.ChatInput()
-agent = cp.AgentComponent()
-url_component = cp.URLComponent()
-tools = await url_component.to_toolkit()
-
-agent.set(
-    model_name="gpt-4o-mini",
-    agent_llm="OpenAI",
-    api_key=os.getenv("OPENAI_API_KEY"),
-    input_value=chat_input.message_response,
-    tools=tools,
-)
-chat_output = cp.ChatOutput().set(input_value=agent.message_response)
-
-graph = Graph(chat_input, chat_output, log_config=log_config)
+
+async def get_graph() -> Graph:
+    """Create and return the graph with async component initialization.
+
+    This function properly handles async component initialization without
+    blocking the module loading process. The script loader will detect this
+    async function and handle it appropriately using run_until_complete.
+
+    Returns:
+        Graph: The configured graph with ChatInput → Agent → ChatOutput flow
+    """
+    log_config = LogConfig(
+        log_level="INFO",
+        log_file=Path("langflow.log"),
+    )
+
+    # Showcase the new flattened component access - no need for deep imports!
+    chat_input = cp.ChatInput()
+    agent = cp.AgentComponent()
+
+    # Use URLComponent for web search capabilities
+    url_component = cp.URLComponent()
+
+    # Properly handle async component initialization
+    tools = await url_component.to_toolkit()
+
+    agent.set(
+        model_name="gpt-4o-mini",
+        agent_llm="OpenAI",
+        api_key=os.getenv("OPENAI_API_KEY"),
+        input_value=chat_input.message_response,
+        tools=tools,
+    )
+    chat_output = cp.ChatOutput().set(input_value=agent.message_response)
+
+    return Graph(chat_input, chat_output, log_config=log_config)
 '''
 
     @pytest.fixture
@@ -99,10 +115,11 @@ def test_agent_script_structure_and_syntax(self, simple_agent_script_content):
         assert "cp.AgentComponent()" in simple_agent_script_content
         assert "cp.URLComponent()" in simple_agent_script_content
         assert "cp.ChatOutput()" in simple_agent_script_content
-        assert "url_component.to_toolkit()" in simple_agent_script_content
+        assert "async def get_graph()" in simple_agent_script_content
+        assert "await url_component.to_toolkit()" in simple_agent_script_content
         assert 'model_name="gpt-4o-mini"' in simple_agent_script_content
         assert 'agent_llm="OpenAI"' in simple_agent_script_content
-        assert "Graph(chat_input, chat_output" in simple_agent_script_content
+        assert "return Graph(chat_input, chat_output" in simple_agent_script_content
 
     def test_agent_script_file_validation(self, simple_agent_script_file):
         """Test that the agent script file exists and has valid content."""
@@ -114,7 +131,8 @@ def test_agent_script_file_validation(self, simple_agent_script_file):
         content = simple_agent_script_file.read_text()
         assert "from lfx import components as cp" in content
         assert "cp.AgentComponent()" in content
-        assert "Graph(chat_input, chat_output" in content
+        assert "async def get_graph()" in content
+        assert "return Graph(chat_input, chat_output" in content
 
     def test_agent_script_supports_formats(self, simple_agent_script_file):
         """Test that the script supports logging configuration."""
@@ -247,15 +265,15 @@ def test_agent_configuration_workflow(self):
         agent.set(
             model_name="gpt-4o-mini",
             agent_llm="OpenAI",
-            api_key="test-key",  # Use test key
+            api_key="test-key",  # pragma: allowlist secret
             input_value="Test message",
             tools=[],  # Empty tools for this test
         )
 
         # Verify configuration was applied
         assert agent.model_name == "gpt-4o-mini"
         assert agent.agent_llm == "OpenAI"
-        assert agent.api_key == "test-key"
+        assert agent.api_key == "test-key"  # pragma: allowlist secret
         assert agent.input_value == "Test message"
 
     def test_chat_output_chaining_pattern(self):

src/lfx/README.md

@@ -159,6 +159,7 @@ Features:
 - Creates an agent with OpenAI GPT model
 - Provides web search tools via URLComponent
 - Connects ChatInput → Agent → ChatOutput
+- Uses async get_graph() function for proper async handling
 
 Usage:
     uv run lfx run simple_agent.py "How are you?"
@@ -172,27 +173,40 @@ from lfx import components as cp
 from lfx.graph import Graph
 from lfx.log.logger import LogConfig
 
-log_config = LogConfig(
-    log_level="INFO",
-    log_file=Path("langflow.log"),
-)
 
-# Showcase the new flattened component access - no need for deep imports!
-chat_input = cp.ChatInput()
-agent = cp.AgentComponent()
-url_component = cp.URLComponent()
-tools = url_component.to_toolkit()
-
-agent.set(
-    model_name="gpt-4.1-mini",
-    agent_llm="OpenAI",
-    api_key=os.getenv("OPENAI_API_KEY"),
-    input_value=chat_input.message_response,
-    tools=tools,
-)
-chat_output = cp.ChatOutput().set(input_value=agent.message_response)
-
-graph = Graph(chat_input, chat_output, log_config=log_config)
+async def get_graph() -> Graph:
+    """Create and return the graph with async component initialization.
+
+    This function properly handles async component initialization without
+    blocking the module loading process. The script loader will detect this
+    async function and handle it appropriately.
+
+    Returns:
+        Graph: The configured graph with ChatInput → Agent → ChatOutput flow
+    """
+    log_config = LogConfig(
+        log_level="INFO",
+        log_file=Path("langflow.log"),
+    )
+
+    # Showcase the new flattened component access - no need for deep imports!
+    chat_input = cp.ChatInput()
+    agent = cp.AgentComponent()
+
+    # Use URLComponent for web search capabilities
+    url_component = cp.URLComponent()
+    tools = await url_component.to_toolkit()
+
+    agent.set(
+        model_name="gpt-4.1-mini",
+        agent_llm="OpenAI",
+        api_key=os.getenv("OPENAI_API_KEY"),
+        input_value=chat_input.message_response,
+        tools=tools,
+    )
+    chat_output = cp.ChatOutput().set(input_value=agent.message_response)
+
+    return Graph(chat_input, chat_output, log_config=log_config)
 ```
 
 **Step 2: Install dependencies**

src/lfx/src/lfx/cli/commands.py

@@ -6,10 +6,12 @@
 import os
 import sys
 import tempfile
+from functools import partial
 from pathlib import Path
 
 import typer
 import uvicorn
+from asyncer import syncify
 from dotenv import load_dotenv
 from rich.console import Console
 from rich.panel import Panel
@@ -32,7 +34,8 @@
 API_KEY_MASK_LENGTH = 8
 
 
-def serve_command(
+@partial(syncify, raise_sync_error=False)
+async def serve_command(
     script_path: str | None = typer.Argument(
         None,
         help=(
@@ -201,12 +204,12 @@ def serve_command(
             raise typer.Exit(1)
 
         if resolved_path.suffix == ".json":
-            graph = load_graph_from_path(resolved_path, resolved_path.suffix, verbose_print, verbose=verbose)
+            graph = await load_graph_from_path(resolved_path, resolved_path.suffix, verbose_print, verbose=verbose)
         elif resolved_path.suffix == ".py":
             verbose_print("Loading graph from Python script...")
             from lfx.cli.script_loader import load_graph_from_script
 
-            graph = load_graph_from_script(resolved_path)
+            graph = await load_graph_from_script(resolved_path)
             verbose_print("✓ Graph loaded from Python script")
         else:
             err_msg = "Error: Only JSON flow files (.json) or Python scripts (.py) are supported. "

src/lfx/src/lfx/cli/common.py

@@ -224,7 +224,7 @@ def validate_script_path(script_path: Path | str, verbose_print) -> tuple[str, P
     return file_extension, script_path
 
 
-def load_graph_from_path(script_path: Path, file_extension: str, verbose_print, *, verbose: bool = False):
+async def load_graph_from_path(script_path: Path, file_extension: str, verbose_print, *, verbose: bool = False):
     """Load a graph from a Python script or JSON file.
 
     Args:
@@ -259,7 +259,7 @@ def load_graph_from_path(script_path: Path, file_extension: str, verbose_print,
                 raise ValueError(error_msg)
 
             verbose_print("Loading graph...")
-            graph = load_graph_from_script(script_path)
+            graph = await load_graph_from_script(script_path)
         else:  # .json
             verbose_print("Loading JSON flow...")
             graph = load_flow_from_json(script_path, disable_logs=not verbose)

src/lfx/src/lfx/cli/run.py

@@ -227,7 +227,7 @@ async def run(
                 typer.echo(f"Type: {graph_info['type']}", file=sys.stderr)
                 typer.echo(f"Source: {graph_info['source_line']}", file=sys.stderr)
                 typer.echo("Loading and executing script...", file=sys.stderr)
-            graph = load_graph_from_script(script_path)
+            graph = await load_graph_from_script(script_path)
         elif file_extension == ".json":
             if verbosity > 0:
                 typer.echo("Valid JSON flow file detected", file=sys.stderr)