feat: agentic UX

[edwinjosechittilappilly]

This pull request introduces several new modules and utilities to support agentic features in Langflow, including component search, template utilities, and support functions. It also adds onboarding and guidance "nudges" to the agentic schema for improved user experience. The changes are grouped into new backend functionalities, schema enhancements, and utility additions.

New backend agentic functionalities:

• Added component_search.py with asynchronous utilities for searching, filtering, and retrieving Langflow components by name, type, and metadata fields, including error handling and documentation.
• Added support.py with a function to normalize data dictionaries by replacing missing or invalid values (None, 'null', NaN, etc.) with a "Not available" string and ensuring required fields are present.

Agentic schema enhancements:

• Introduced schema.json defining a set of "nudges"—onboarding prompts and quick actions for the Langflow Assistant UI, such as navigation to onboarding, creating flows, browsing components, exploring templates, reading docs, and sharing feedback.

Utility and module structure improvements:

• Added utils/__init__.py to expose template search utilities for agentic features, including functions for listing templates and retrieving tags/counts.
• Added module docstrings and initialization files for langflow.agentic and its submodules to clarify their purpose and improve maintainability. [1] [2]

Entrypoint for MCP server:

• Added mcp/__main__.py to provide a Python module entrypoint for running the Langflow Agentic MCP server via python -m langflow.agentic.mcp.

Summary by CodeRabbit

• New Features

  • Added AI-powered agentic server with tools to search templates, discover components, and inspect flows.
  • Introduced Assistant button for AI-assisted field value suggestions in the UI.
  • Created Langflow Assistant folder for pre-built flows and enhanced onboarding.

• Chores

  • Auto-configured agentic features during application startup.
  • Enhanced template and component discovery utilities.

[edwinjosechittilappilly]

feature flagged version of Agentic UX @Cristhianzl @lucaseduoli
Merging since PR was getting larger.

[edwinjosechittilappilly]

test_dynamic_module_discovery passes locally but fails in CI.

[edwinjosechittilappilly]

Backend Fix: #10636

[codeflash-ai]

⚡️Codeflash found 31% (0.31x) speedup for replace_none_and_null_with_empty_str in src/backend/base/langflow/agentic/mcp/support.py

⏱️ Runtime : 3.04 milliseconds → 2.31 milliseconds (best of 155 runs)

📝 Explanation and details

The optimized code achieves a 31% speedup through several key performance improvements:

1. Eliminated Redundant Set Creation
The original code created set(required_fields) if required_fields else set() on every function call, even when required_fields was None. The optimization changes this to set(required_fields) if required_fields else None, avoiding unnecessary empty set creation.

2. Branch Optimization for Required Fields
The most significant improvement comes from splitting the main loop into two branches based on whether required_fields_set exists. This eliminates the expensive required_fields_set - new_dict.keys() set difference operation for cases where no required fields are specified (which appears to be ~67% of the test data based on the profiler showing 2741 iterations in the "else" branch vs 511 in the "if" branch).

3. Precomputed Constants
Moving null_strs = {"null", "nan", "infinity", "-infinity"} and not_avail = "Not available" outside the function definitions prevents repeated allocations during each call.

4. Improved Type Checking
Using elif isinstance(v, float) instead of separate if statements ensures mutually exclusive type checks, providing marginal performance gains by avoiding unnecessary isinstance calls.

5. Optimized Attribute Access
The hasattr(v, "isnat") check is now separated from the getattr call, avoiding the redundant attribute lookup when the attribute doesn't exist.

Performance Impact Analysis:

• The line profiler shows the most expensive operation (new_dict = {k: convert_value(v) for k, v in d.items()}) remains the bottleneck at ~64% of total time
• However, the set difference operation (required_fields_set - new_dict.keys()) was reduced from 6.8% to 1.3% of total time
• The optimization is particularly effective for workloads where required_fields is frequently None or empty, as evidenced by the large "else" branch in the profiler results

These optimizations maintain identical behavior while significantly reducing computational overhead, especially beneficial for data processing pipelines that frequently call this function with varying required_fields parameters.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 51 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import math

imports

import pytest
from langflow.agentic.mcp.support import replace_none_and_null_with_empty_str

unit tests

1. Basic Test Cases

def test_basic_none_and_null_replacement():
# Test replacing None and 'null' (case-insensitive) values with "Not available"
data = [
{"a": None, "b": "null", "c": "NuLl", "d": 1, "e": "something"},
{"a": "NULL", "b": 2, "c": None, "d": "ok", "e": "null"}
]
expected = [
{"a": "Not available", "b": "Not available", "c": "Not available", "d": 1, "e": "something"},
{"a": "Not available", "b": 2, "c": "Not available", "d": "ok", "e": "Not available"}
]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_basic_nan_and_infinity_strings():
# Test replacing string values 'NaN', 'Infinity', '-Infinity' (case-insensitive) with "Not available"
data = [
{"a": "NaN", "b": "Infinity", "c": "-Infinity", "d": "nan", "e": "infinity", "f": "-infinity"},
{"a": "ok", "b": "good"}
]
expected = [
{"a": "Not available", "b": "Not available", "c": "Not available", "d": "Not available", "e": "Not available", "f": "Not available"},
{"a": "ok", "b": "good"}
]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_basic_float_nan():
# Test replacing float('nan') with "Not available"
data = [
{"a": float('nan'), "b": 5.0, "c": None},
{"a": 1.0, "b": float('nan')}
]
expected = [
{"a": "Not available", "b": 5.0, "c": "Not available"},
{"a": 1.0, "b": "Not available"}
]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_basic_no_replacements_needed():
# Test when no replacements are needed
data = [
{"a": 1, "b": "hello", "c": 2.5},
{"a": "world", "b": 0}
]
expected = [
{"a": 1, "b": "hello", "c": 2.5},
{"a": "world", "b": 0}
]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_basic_required_fields_addition():
# Test required_fields adds missing fields as "Not available"
data = [
{"a": 1, "b": None},
{"a": 2}
]
required_fields = ["a", "b", "c"]
expected = [
{"a": 1, "b": "Not available", "c": "Not available"},
{"a": 2, "b": "Not available", "c": "Not available"}
]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields)

2. Edge Test Cases

def test_edge_empty_input():
# Test with empty input list
data = []
expected = []
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_edge_empty_dicts():
# Test with list of empty dicts
data = [{} for _ in range(3)]
expected = [{} for _ in range(3)]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_edge_non_dict_elements():
# Test with elements in list that are not dicts
data = [
{"a": None},
42,
"not a dict",
{"b": "null"}
]
expected = [
{"a": "Not available"},
42,
"not a dict",
{"b": "Not available"}
]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_edge_mixed_types():
# Test with various types, including booleans, lists, tuples, dicts, objects
class Dummy:
pass
dummy = Dummy()
data = [
{"a": True, "b": False, "c": [None, "null"], "d": (1, None), "e": dummy}
]
expected = [
{"a": True, "b": False, "c": [None, "null"], "d": (1, None), "e": dummy}
]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_edge_isnat_attribute():
# Simulate pandas.NaT-like object
class FakeNaT:
isnat = True
data = [{"a": FakeNaT()}]
expected = [{"a": "Not available"}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_edge_required_fields_superset():
# required_fields contains fields not in any dict
data = [{"x": 1}]
required_fields = ["x", "y", "z"]
expected = [{"x": 1, "y": "Not available", "z": "Not available"}]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields)

def test_edge_required_fields_empty():
# required_fields is empty
data = [{"a": None}]
required_fields = []
expected = [{"a": "Not available"}]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields)

def test_edge_whitespace_and_case():
# Test handling of whitespace and case in string values
data = [
{"a": " Null ", "b": " nAn", "c": " -INFINITY ", "d": "Infinity "}
]
expected = [
{"a": "Not available", "b": "Not available", "c": "Not available", "d": "Not available"}
]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_edge_required_fields_on_non_dict():
# If required_fields is provided but element is not a dict, should not error
data = [42, "foo", {"a": None}]
required_fields = ["a", "b"]
expected = [42, "foo", {"a": "Not available", "b": "Not available"}]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields)

def test_edge_dict_with_non_string_keys():
# Dict with integer keys, should not break
data = [{1: None, 2: "null", "a": "nan"}]
expected = [{1: "Not available", 2: "Not available", "a": "Not available"}]
codeflash_output = replace_none_and_null_with_empty_str(data)

3. Large Scale Test Cases

def test_large_many_dicts():
# Test with 1000 dicts, each with a mix of values
data = []
for i in range(1000):
d = {
"a": None if i % 2 == 0 else i,
"b": "null" if i % 3 == 0 else str(i),
"c": float('nan') if i % 5 == 0 else i * 0.1,
"d": "ok"
}
data.append(d)
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output

def test_large_required_fields():
# Test with large required_fields list
data = [{"a": None, "b": 1}]
required_fields = [chr(97 + i) for i in range(20)] # ['a', ..., 't']
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields); result = codeflash_output
d = result[0]
for k in required_fields:
pass
for k in required_fields:
if k not in {"a", "b"}:
pass

def test_large_all_non_dicts():
# Test with a large list of non-dict elements
data = [i for i in range(1000)]
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output

def test_large_mixed_dicts_and_non_dicts():
# Test with alternating dict and non-dict entries
data = []
for i in range(500):
data.append({"a": None, "b": i})
data.append(i)
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output
for i in range(500):
pass

codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

#------------------------------------------------
import math

imports

import pytest
from langflow.agentic.mcp.support import replace_none_and_null_with_empty_str

unit tests

----------------------

Basic Test Cases

----------------------

def test_empty_list_returns_empty():
# Test with empty list input
codeflash_output = replace_none_and_null_with_empty_str([])

def test_single_dict_no_none_or_null():
# Dict with only valid values
data = [{"a": 1, "b": "hello"}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_single_dict_with_none():
# Dict with None value
data = [{"a": None, "b": "test"}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_single_dict_with_null_string_variants():
# Dict with various 'null' string forms
data = [{"a": "null", "b": "NuLl", "c": "NULL", "d": " null ", "e": "nUll"}]
expected = {k: "Not available" for k in data[0].keys()}
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_single_dict_with_nan_float():
# Dict with float('nan')
data = [{"a": float('nan'), "b": 2}]
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output

def test_single_dict_with_nan_string():
# Dict with string 'nan' (case-insensitive)
data = [{"a": "nan", "b": "NaN", "c": " NAN "}]
expected = {k: "Not available" for k in data[0].keys()}
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_single_dict_with_infinity_variants():
# Dict with 'infinity', '-infinity' as strings (case-insensitive)
data = [{"a": "infinity", "b": "Infinity", "c": "INFINITY", "d": " infinity ", "e": "-infinity", "f": "-Infinity"}]
expected = {k: "Not available" for k in data[0].keys()}
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_single_dict_with_regular_values():
# Dict with normal values, should remain unchanged
data = [{"a": 123, "b": "foo", "c": False, "d": 0.0}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_multiple_dicts_mixed_values():
# List of dicts with mixed values
data = [
{"a": None, "b": "null"},
{"a": "hello", "b": 42},
{"a": float('nan'), "b": "NaN"},
]
expected = [
{"a": "Not available", "b": "Not available"},
{"a": "hello", "b": 42},
{"a": "Not available", "b": "Not available"},
]
codeflash_output = replace_none_and_null_with_empty_str(data)

----------------------

Edge Test Cases

----------------------

def test_dict_with_missing_required_fields():
# Dict missing required fields, should add them with "Not available"
data = [{"a": 1}]
required_fields = ["a", "b", "c"]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields); result = codeflash_output

def test_dict_with_all_required_fields_present():
# Dict has all required fields, none should be added
data = [{"a": 1, "b": 2, "c": 3}]
required_fields = ["a", "b", "c"]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields); result = codeflash_output

def test_dict_with_fields_all_none_and_missing_required():
# All present fields are None, and some required fields missing
data = [{"a": None}]
required_fields = ["a", "b"]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields); result = codeflash_output

def test_dict_with_nested_dicts_are_untouched():
# Nested dicts should not be altered
data = [{"a": {"b": None, "c": "null"}}]
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output

def test_non_dict_in_data_list():
# Non-dict elements should be returned as-is
data = [{"a": None}, 123, "string", None, [1, 2, 3]]
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output

def test_empty_dict_in_data_list():
# Empty dict should be handled, and required fields filled
data = [{}]
required_fields = ["x", "y"]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields); result = codeflash_output

def test_dict_with_custom_object_with_isnat():
# Custom object with isnat attribute True should be replaced
class FakeNaT:
isnat = True
data = [{"a": FakeNaT()}]
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output

def test_dict_with_custom_object_with_isnat_false():
# Custom object with isnat attribute False should not be replaced
class NotNaT:
isnat = False
obj = NotNaT()
data = [{"a": obj}]
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output

def test_required_fields_empty_list():
# required_fields as empty list should not add any fields
data = [{"a": 1}]
codeflash_output = replace_none_and_null_with_empty_str(data, []); result = codeflash_output

def test_required_fields_none():
# required_fields as None should not add any fields
data = [{"a": 1}]
codeflash_output = replace_none_and_null_with_empty_str(data, None); result = codeflash_output

def test_required_fields_with_overlapping_and_new_fields():
# required_fields overlapping with present and missing fields
data = [{"a": 1, "b": None}]
required_fields = ["a", "b", "c"]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields); result = codeflash_output

def test_dict_with_whitespace_strings():
# Whitespace-only strings should not be replaced
data = [{"a": " ", "b": "\n\t"}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_dict_with_numeric_strings():
# Numeric strings should not be replaced
data = [{"a": "123", "b": "0.0", "c": "-1"}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_dict_with_float_infinity():
# float('inf') and float('-inf') should not be replaced
data = [{"a": float('inf'), "b": float('-inf')}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_dict_with_mixed_types():
# Dict with various types
data = [{"a": None, "b": "null", "c": float('nan'), "d": 42, "e": "foo"}]
expected = {"a": "Not available", "b": "Not available", "c": "Not available", "d": 42, "e": "foo"}
codeflash_output = replace_none_and_null_with_empty_str(data)

----------------------

Large Scale Test Cases

----------------------

def test_large_list_of_dicts():
# Test with 1000 dicts, each with some None and 'null'
data = [{"a": None if i % 2 == 0 else "null", "b": i} for i in range(1000)]
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output
for i, d in enumerate(result):
pass

def test_large_list_with_required_fields():
# 500 dicts, each missing some required fields
data = [{"x": i} for i in range(500)]
required_fields = ["x", "y", "z"]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields); result = codeflash_output
for i, d in enumerate(result):
pass

def test_large_dicts_with_various_nulls():
# 100 dicts, each with 10 fields, some None, some 'null', some valid
data = []
for i in range(100):
d = {}
for j in range(10):
if j % 3 == 0:
d[f"f{j}"] = None
elif j % 3 == 1:
d[f"f{j}"] = "null"
else:
d[f"f{j}"] = i * j
data.append(d)
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output
for d in result:
for j in range(10):
key = f"f{j}"
if j % 3 in (0, 1):
pass
else:
pass

def test_large_list_with_non_dicts():
# List with 100 dicts and 100 non-dict elements
data = [{"a": None} for _ in range(100)] + [42] * 100
codeflash_output = replace_none_and_null_with_empty_str(data); result = codeflash_output
for i in range(100):
pass
for i in range(100, 200):
pass

----------------------

Mutation Testing Guards

----------------------

def test_mutation_guard_null_string_must_be_case_insensitive():
# If 'null' is not checked case-insensitively, this test will fail
data = [{"a": "NuLl"}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_mutation_guard_nan_float_must_be_checked():
# If math.isnan is not used, this test will fail
data = [{"a": float('nan')}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_mutation_guard_none_must_be_replaced():
# If None is not replaced, this test will fail
data = [{"a": None}]
codeflash_output = replace_none_and_null_with_empty_str(data)

def test_mutation_guard_required_fields_added():
# If required_fields are not added, this test will fail
data = [{"a": 1}]
required_fields = ["a", "b"]
codeflash_output = replace_none_and_null_with_empty_str(data, required_fields)

codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To test or edit this optimization locally git merge codeflash/optimize-pr10567-2025-11-18T15.28.14

Click to see suggested changes

Suggested change

	def convert_value(v):
	if v is None:
	return "Not available"
	if isinstance(v, str):
	v_stripped = v.strip().lower()
	if v_stripped in {"null", "nan", "infinity", "-infinity"}:
	return "Not available"
	if isinstance(v, float):
	try:
	if math.isnan(v):
	return "Not available"
	except Exception as e: # noqa: BLE001
	logger.aexception(f"Error converting value {v} to float: {e}")
	if hasattr(v, "isnat") and getattr(v, "isnat", False):
	return "Not available"
	return v
	not_avail = "Not available"
	required_fields_set = set(required_fields) if required_fields else set()
	result = []
	for d in data:
	if not isinstance(d, dict):
	result.append(d)
	continue
	new_dict = {k: convert_value(v) for k, v in d.items()}
	missing = required_fields_set - new_dict.keys()
	if missing:
	for k in missing:
	new_dict[k] = not_avail
	result.append(new_dict)
	# Precompute sets and string
	null_strs = {"null", "nan", "infinity", "-infinity"}
	not_avail = "Not available"
	required_fields_set = set(required_fields) if required_fields else None
	def convert_value(v):
	if v is None:
	return not_avail
	if isinstance(v, str):
	v_stripped = v.strip().lower()
	if v_stripped in null_strs:
	return not_avail
	elif isinstance(v, float):
	try:
	if math.isnan(v):
	return not_avail
	except Exception as e: # noqa: BLE001
	logger.aexception(f"Error converting value {v} to float: {e}")
	# Optimization: avoid getattr if not needed
	elif hasattr(v, "isnat"):
	isnat = getattr(v, "isnat", False)
	if isnat:
	return not_avail
	return v
	# Use list comprehension for fast iteration and avoid repeated set()-construction
	result: list[dict] = []
	if required_fields_set:
	for d in data:
	if not isinstance(d, dict):
	result.append(d)
	continue
	new_dict = {k: convert_value(v) for k, v in d.items()}
	# Only run if required_fields_set is not empty
	missing = required_fields_set - new_dict.keys()
	if missing:
	# Use dict.update for efficiency when adding more than one key
	for k in missing:
	new_dict[k] = not_avail
	result.append(new_dict)
	else:
	# Avoid computing missing keys when not needed
	for d in data:
	if not isinstance(d, dict):
	result.append(d)
	continue
	new_dict = {k: convert_value(v) for k, v in d.items()}
	result.append(new_dict)

[codeflash-ai]

⚡️Codeflash found 90% (0.90x) speedup for list_all_tags in src/backend/base/langflow/agentic/mcp/server.py

⏱️ Runtime : 383 milliseconds → 201 milliseconds (best of 17 runs)

📝 Explanation and details

The key optimization is replacing text-based file reading with binary reading for JSON parsing. The original code used Path(template_file).read_text(encoding="utf-8") which forces UTF-8 decoding, while the optimized version uses template_file.read_bytes() directly.

What changed:

1. File I/O optimization: read_text(encoding="utf-8") → read_bytes()
2. Removed redundant Path construction: Path(template_file) → template_file (since glob already returns Path objects)
3. Streamlined exception handling: Only catch orjson.JSONDecodeError instead of both json.JSONDecodeError and orjson.JSONDecodeError

Why it's faster:
The performance gain comes from avoiding UTF-8 decoding overhead. When using read_text(), Python must:

1. Read file bytes
2. Decode bytes to UTF-8 string
3. Pass string to orjson.loads()

With read_bytes(), orjson can parse directly from bytes, eliminating the intermediate string conversion. The line profiler shows the critical file reading operation dropped from ~409ms to ~205ms (50% reduction), which explains most of the overall 90% speedup.

Impact on workloads:
This optimization particularly benefits scenarios with many JSON files or large JSON content, as seen in the test cases with 100+ files or 1000+ tags. The function appears to be used for template discovery in the MCP server context, so faster startup times when scanning template directories would be noticeable to users.

Test case performance:
The optimization excels with larger datasets - tests with hundreds of files or extensive tag collections benefit most from the reduced I/O overhead per file operation.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 22 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Generated Regression Tests and Runtime

import json
import os
import shutil
import tempfile
from pathlib import Path

imports

import pytest
from langflow.agentic.mcp.server import list_all_tags

Helper fixture to create a temporary starter_projects directory for tests

@pytest.fixture
def starter_projects_dir():
temp_dir = tempfile.mkdtemp()
yield temp_dir
shutil.rmtree(temp_dir)

Helper to create a template file in the directory

def create_template_file(directory, filename, tags=None, extra=None):
data = {}
if tags is not None:
data["tags"] = tags
if extra is not None:
data.update(extra)
file_path = os.path.join(directory, filename)
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f)
return file_path

Global variable to pass path to list_all_tags

TEST_STARTER_PROJECTS_PATH = None

----------- BASIC TEST CASES -----------

def test_list_all_tags_single_file_single_tag(starter_projects_dir):
"""Test: Single file with a single tag."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["agents"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_multiple_files_unique_tags(starter_projects_dir):
"""Test: Multiple files, each with unique tags."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["agents"])
create_template_file(starter_projects_dir, "template2.json", tags=["chatbots"])
create_template_file(starter_projects_dir, "template3.json", tags=["tools"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_multiple_files_overlapping_tags(starter_projects_dir):
"""Test: Multiple files with overlapping tags."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["agents", "tools"])
create_template_file(starter_projects_dir, "template2.json", tags=["tools", "chatbots"])
create_template_file(starter_projects_dir, "template3.json", tags=["chatbots", "rag"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_tags_are_unsorted(starter_projects_dir):
"""Test: Tags in files are unsorted; output should be sorted."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["zeta", "alpha", "gamma"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_tags_with_duplicates_in_file(starter_projects_dir):
"""Test: Tags in a single file contain duplicates."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["agents", "agents", "tools"])
codeflash_output = list_all_tags(); result = codeflash_output

----------- EDGE TEST CASES -----------

def test_list_all_tags_no_files(starter_projects_dir):
"""Test: Directory contains no JSON files."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_file_with_no_tags_key(starter_projects_dir):
"""Test: File missing 'tags' key."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=None, extra={"name": "foo"})
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_file_with_tags_empty_list(starter_projects_dir):
"""Test: File with 'tags' as an empty list."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=[])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_file_with_tags_not_a_list(starter_projects_dir):
"""Test: File with 'tags' as a string (invalid type)."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
# Write a file with "tags": "agents"
file_path = os.path.join(starter_projects_dir, "template1.json")
with open(file_path, "w", encoding="utf-8") as f:
json.dump({"tags": "agents"}, f)
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_file_with_tags_none(starter_projects_dir):
"""Test: File with 'tags' as None."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=None)
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_file_with_invalid_json(starter_projects_dir):
"""Test: File with invalid JSON format."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
file_path = os.path.join(starter_projects_dir, "bad_template.json")
# Write invalid JSON
with open(file_path, "w", encoding="utf-8") as f:
f.write("{invalid json")
# Should not raise, should skip invalid file
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_non_json_files_ignored(starter_projects_dir):
"""Test: Non-JSON files are ignored."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
# Create a .txt file
with open(os.path.join(starter_projects_dir, "not_a_json.txt"), "w", encoding="utf-8") as f:
f.write("tags: agents")
# No .json files, so result should be empty
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_tags_with_special_characters(starter_projects_dir):
"""Test: Tags with special characters and unicode."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["αβγ", "tools!", "chat bots"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_tags_are_case_sensitive(starter_projects_dir):
"""Test: Tags with different casing are treated as distinct."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["Agents", "agents", "AGENTS"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_tags_with_empty_strings(starter_projects_dir):
"""Test: Tags with empty string."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["", "agents"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_tags_with_none_in_list(starter_projects_dir):
"""Test: Tags list contains None values."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
create_template_file(starter_projects_dir, "template1.json", tags=["agents", None, "tools"])
codeflash_output = list_all_tags(); result = codeflash_output
# None is not a string, but is hashable and sortable; will be included
# But sorting with None and str raises TypeError, so function should ignore None
# Our implementation does not filter out None, so it will raise TypeError
# To match the original, we should filter out non-str tags
# Let's update get_all_tags for this edge case:
# all_tags.update([t for t in tags if isinstance(t, str)])
# For now, we expect TypeError, so we catch it and assert
try:
_ = result
except TypeError:
pass

----------- LARGE SCALE TEST CASES -----------

def test_list_all_tags_large_number_of_files_and_tags(starter_projects_dir):
"""Test: Large number of files and tags."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
# Create 100 files, each with 10 unique tags
all_tags = set()
for i in range(100):
tags = [f"tag_{i}{j}" for j in range(10)]
all_tags.update(tags)
create_template_file(starter_projects_dir, f"template{i}.json", tags=tags)
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_large_number_of_files_some_with_no_tags(starter_projects_dir):
"""Test: Large number of files, some with no tags."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
all_tags = set()
# 50 files with tags, 50 without
for i in range(50):
tags = [f"tag_{i}{j}" for j in range(5)]
all_tags.update(tags)
create_template_file(starter_projects_dir, f"template_with_tags{i}.json", tags=tags)
for i in range(50):
create_template_file(starter_projects_dir, f"template_without_tags_{i}.json", tags=None)
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_large_number_of_files_with_duplicate_tags(starter_projects_dir):
"""Test: Large number of files, many duplicate tags."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
# 200 files, all with the same tag
for i in range(200):
create_template_file(starter_projects_dir, f"template_{i}.json", tags=["shared_tag"])
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_performance_under_1000_tags(starter_projects_dir):
"""Test: Performance and correctness with nearly 1000 tags."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
tags = [f"tag_{i}" for i in range(999)]
create_template_file(starter_projects_dir, "template1.json", tags=tags)
codeflash_output = list_all_tags(); result = codeflash_output

def test_list_all_tags_performance_under_1000_files(starter_projects_dir):
"""Test: Performance and correctness with nearly 1000 files."""
global TEST_STARTER_PROJECTS_PATH
TEST_STARTER_PROJECTS_PATH = starter_projects_dir
# Each file has a unique tag
for i in range(999):
create_template_file(starter_projects_dir, f"template_{i}.json", tags=[f"tag_{i}"])
codeflash_output = list_all_tags(); result = codeflash_output
expected_tags = [f"tag_{i}" for i in range(999)]

codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

#------------------------------------------------
import json
import os
import shutil
import tempfile
from pathlib import Path

imports

import pytest
from langflow.agentic.mcp.server import list_all_tags

--- Unit tests ---

Helper to create a temporary starter_projects directory with given files

@pytest.fixture
def temp_starter_projects(tmp_path):
"""Creates a temporary starter_projects directory for tests."""
dir_path = tmp_path / "starter_projects"
dir_path.mkdir()
return dir_path

---------- BASIC TEST CASES ----------

def test_file_with_tags_as_none(temp_starter_projects):
"""Test with a file where 'tags' is None."""
file = temp_starter_projects / "template1.json"
file.write_text(json.dumps({"tags": None}), encoding="utf-8")
# None is not iterable, so update() will raise TypeError
# Let's catch it and assert that the function raises
with pytest.raises(TypeError):
list_all_tags(temp_starter_projects)

To test or edit this optimization locally git merge codeflash/optimize-pr10567-2025-11-19T21.50.18

Suggested change

	template_data = orjson.loads(Path(template_file).read_text(encoding="utf-8"))
	tags = template_data.get("tags", [])
	all_tags.update(tags)
	except (json.JSONDecodeError, orjson.JSONDecodeError) as e:
	template_data = orjson.loads(template_file.read_bytes())
	tags = template_data.get("tags", [])
	all_tags.update(tags)
	except orjson.JSONDecodeError as e:

[Adam-Aghili]

@edwinjosechittilappilly the frontend tests are flaky so if we re run it will eventually pass. I don't know if we want to address that in this issue. Even with the wait for load sometiems the tools is not available. we could also increase the retries but that would increase our already long CI time