diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Nvidia Remix.json b/src/backend/base/langflow/initial_setup/starter_projects/Nvidia Remix.json index 1d68d82d8a..37b7e190aa 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Nvidia Remix.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Nvidia Remix.json @@ -2097,6 +2097,7 @@ "use_cache", "verify_ssl", "headers", + "tool_execution_timeout", "tool", "tool_placeholder" ], @@ -2107,7 +2108,7 @@ "legacy": false, "lf_version": "1.4.2", "metadata": { - "code_hash": "94824f26f31e", + "code_hash": "c7652a63f23e", "dependencies": { "dependencies": [ { @@ -2173,7 +2174,7 @@ "show": true, "title_case": false, "type": "code", - "value": "from __future__ import annotations\n\nimport asyncio\nimport hashlib\nimport json\nimport time\nimport uuid\nfrom types import UnionType\nfrom typing import Any, get_args, get_origin\n\nfrom langchain_core.tools import StructuredTool # noqa: TC002\nfrom pydantic import BaseModel\n\nfrom lfx.base.agents.utils import maybe_unflatten_dict, safe_cache_get, safe_cache_set\nfrom lfx.base.mcp.util import (\n MCPStdioClient,\n MCPStreamableHttpClient,\n update_tools,\n)\nfrom lfx.base.tools.constants import TOOL_OUTPUT_DISPLAY_NAME, TOOL_OUTPUT_NAME\nfrom lfx.custom.custom_component.component_with_cache import ComponentWithCache\nfrom lfx.inputs.inputs import InputTypes # noqa: TC001\nfrom lfx.io import BoolInput, DictInput, DropdownInput, McpInput, MessageTextInput, Output\nfrom lfx.io.schema import schema_to_langflow_inputs\nfrom lfx.log.logger import logger\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.services.deps import get_storage_service, session_scope\n\n\ndef resolve_mcp_config(\n server_name: str, # noqa: ARG001\n server_config_from_value: dict | None,\n server_config_from_db: dict | None,\n) -> dict | None:\n \"\"\"Resolve MCP server config with proper precedence.\n\n Resolves the configuration for an MCP server with the following precedence:\n 1. Database config (takes priority) - ensures edits are reflected\n 2. Config from value/tweaks (fallback) - allows REST API to provide config for new servers\n\n Args:\n server_name: Name of the MCP server\n server_config_from_value: Config provided via value/tweaks (optional)\n server_config_from_db: Config from database (optional)\n\n Returns:\n Final config to use (DB takes priority, falls back to value)\n Returns None if no config found in either location\n \"\"\"\n if server_config_from_db:\n return server_config_from_db\n return server_config_from_value\n\n\n# TODO(legacy-cleanup): This file is ~800 lines, over the 500-line guideline. Split\n# helper functions (resolve_mcp_config, connection resolution) from the MCPToolsComponent\n# orchestration logic in a follow-up PR.\nclass MCPToolsComponent(ComponentWithCache):\n \"\"\"MCP Tools component.\n\n Behaviour notes:\n - Stale agent tools vs server were caused by Langflow caching the Toolset output\n (``Output.cache=True`` plus the persisted ``output.value`` in saved flow JSON), not\n only by the user-facing \"Use Cached Server\" (``use_cache``) toggle. ``_build_tool_output``\n declares the Toolset output with ``cache=False`` and ``map_outputs`` overrides the\n persisted value for existing flows so every run resolves a fresh tool list.\n - ``update_tool_list`` is serialized with an ``asyncio.Lock`` so concurrent calls do not\n share the same Streamable HTTP client session; overlapping POST/DELETE cycles otherwise\n surface as HTTP 404 and the MCP SDK reports ``Session terminated``.\n - All diagnostic logs are at DEBUG level to avoid swamping logs on hot paths (an agent\n may call ``_get_tools`` per step). Header values are never logged; only header *keys*.\n \"\"\"\n\n # Short-lived in-memory cache window (seconds) for ``_get_tools``. When the\n # same header-hash key is requested again within this window the cached tool\n # list is reused so parallel agent runs that share identical auth don't\n # each pay for a fresh MCP round-trip. Set to ``0`` to disable. This is\n # distinct from the \"Use Cached Server\" (``use_cache``) toggle which only\n # controls the shared cross-request server cache.\n TOOL_TTL_SECS: int = 30\n # Upper bound on the per-instance TTL cache. Oldest entries are evicted\n # when the cap is reached. Keeps memory bounded for flows where the same\n # component handles many rotating auth contexts (e.g. per-tenant tokens).\n TOOL_TTL_MAX_ENTRIES: int = 32\n # Upper bound on the shared cross-request ``servers`` cache. Each\n # (server_name, header-hash) pair is a distinct entry; without a bound a\n # tenant that rotates session tokens would grow this map without limit.\n SHARED_SERVERS_CACHE_MAX_ENTRIES: int = 64\n\n schema_inputs: list = []\n tools: list[StructuredTool] = []\n _not_load_actions: bool = False\n _tool_cache: dict = {}\n _last_selected_server: str | None = None # Cache for the last selected server\n\n def __init__(self, **data) -> None:\n super().__init__(**data)\n # Initialize cache keys to avoid CacheMiss when accessing them\n self._ensure_cache_structure()\n\n # Initialize clients with access to the component cache\n self.stdio_client: MCPStdioClient = MCPStdioClient(component_cache=self._shared_component_cache)\n self.streamable_http_client: MCPStreamableHttpClient = MCPStreamableHttpClient(\n component_cache=self._shared_component_cache\n )\n # One MCP stdio/streamable client pair per component; concurrent update_tool_list calls\n # otherwise race (session DELETE vs POST) and the MCP SDK surfaces HTTP 404 as \"Session terminated\".\n self._update_tool_list_lock = asyncio.Lock()\n # Per-instance TTL cache for ``_get_tools``: {cache_key: (monotonic_ts, tools)}.\n # Declared here (not at class scope) so every component gets its own dict —\n # a class-level dict would be shared across every MCPToolsComponent in the process,\n # leaking tool lists across tenants that happen to hash to the same key.\n self._ttl_tool_cache: dict[str, tuple[float, list]] = {}\n\n def _ensure_cache_structure(self):\n \"\"\"Ensure the cache has the required structure.\"\"\"\n # Check if servers key exists and is not CacheMiss\n servers_value = safe_cache_get(self._shared_component_cache, \"servers\")\n if servers_value is None:\n safe_cache_set(self._shared_component_cache, \"servers\", {})\n\n # Check if last_selected_server key exists and is not CacheMiss\n last_server_value = safe_cache_get(self._shared_component_cache, \"last_selected_server\")\n if last_server_value is None:\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", \"\")\n\n def _normalized_headers_for_cache(self) -> dict[str, str]:\n \"\"\"Component headers as a dict for stable cache keying (auth / tweaks).\"\"\"\n component_headers = getattr(self, \"headers\", None) or []\n if isinstance(component_headers, list):\n return {\n str(item[\"key\"]): str(item[\"value\"])\n for item in component_headers\n if isinstance(item, dict) and \"key\" in item and \"value\" in item\n }\n if isinstance(component_headers, dict):\n return {str(k): str(v) for k, v in component_headers.items()}\n return {}\n\n def _mcp_servers_cache_key(self, server_name: str) -> str:\n \"\"\"Cache key for shared servers map; includes headers so auth/tweak changes get distinct entries.\"\"\"\n if not server_name:\n return \"\"\n hdrs = self._normalized_headers_for_cache()\n if not hdrs:\n return server_name\n payload = json.dumps(hdrs, sort_keys=True)\n digest = hashlib.sha256(payload.encode()).hexdigest()[:16]\n return f\"{server_name}:{digest}\"\n\n def _build_tool_output(self) -> Output:\n # Do not cache Toolset output. This is separate from the MCP \"Use Cached Server\" (use_cache)\n # toggle: Langflow's Output.cache defaults to True and was memoizing the first to_toolkit()\n # result, so per-request tweaks/headers never refreshed bound tools even when use_cache=False.\n return Output(\n name=TOOL_OUTPUT_NAME,\n display_name=TOOL_OUTPUT_DISPLAY_NAME,\n method=\"to_toolkit\",\n types=[\"Tool\"],\n cache=False,\n )\n\n def map_outputs(self) -> None:\n \"\"\"Override the persisted ``component_as_tool`` cache flag from saved flow JSON.\n\n ``_build_tool_output`` already returns the output with ``cache=False``, but the\n flow JSON for existing flows often stores ``cache: true`` for ``component_as_tool``\n and that persisted value wins over the declaration. Forcing ``cache=False`` here\n guarantees saved flows also bypass Output memoization and get a fresh tool list\n on every run.\n \"\"\"\n super().map_outputs()\n if TOOL_OUTPUT_NAME in self._outputs_map:\n self._outputs_map[TOOL_OUTPUT_NAME].cache = False\n\n default_keys: list[str] = [\n \"code\",\n \"_type\",\n \"tool_mode\",\n \"tool_placeholder\",\n \"mcp_server\",\n \"tool\",\n \"use_cache\",\n \"verify_ssl\",\n \"headers\",\n ]\n\n display_name = \"MCP Tools\"\n description = \"Connect to an MCP server to use its tools.\"\n documentation: str = \"https://docs.langflow.org/mcp-tools\"\n icon = \"Mcp\"\n name = \"MCPTools\"\n\n inputs = [\n McpInput(\n name=\"mcp_server\",\n display_name=\"MCP Server\",\n info=\"Select the MCP Server that will be used by this component\",\n real_time_refresh=True,\n ),\n BoolInput(\n name=\"use_cache\",\n display_name=\"Use Cached Server\",\n info=(\n \"Enable caching of MCP Server and tools to improve performance. \"\n \"Disable to always fetch fresh tools and server updates.\"\n ),\n value=False,\n advanced=True,\n ),\n BoolInput(\n name=\"verify_ssl\",\n display_name=\"Verify SSL Certificate\",\n info=(\n \"Enable SSL certificate verification for HTTPS connections. \"\n \"Disable only for development/testing with self-signed certificates.\"\n ),\n value=True,\n advanced=True,\n ),\n DictInput(\n name=\"headers\",\n display_name=\"Headers\",\n info=(\n \"HTTP headers to include with MCP server requests. \"\n \"Useful for authentication (e.g., Authorization header). \"\n \"These headers override any headers configured in the MCP server settings.\"\n ),\n advanced=True,\n is_list=True,\n ),\n DropdownInput(\n name=\"tool\",\n display_name=\"Tool\",\n options=[],\n value=\"\",\n info=\"Select the tool to execute\",\n show=False,\n required=True,\n real_time_refresh=True,\n refresh_button=True,\n ),\n MessageTextInput(\n name=\"tool_placeholder\",\n display_name=\"Tool Placeholder\",\n info=\"Placeholder for the tool\",\n value=\"\",\n show=False,\n tool_mode=False,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Response\", name=\"response\", method=\"build_output\"),\n ]\n\n async def _validate_schema_inputs(self, tool_obj) -> list[InputTypes]:\n \"\"\"Validate and process schema inputs for a tool.\"\"\"\n try:\n if not tool_obj or not hasattr(tool_obj, \"args_schema\"):\n msg = \"Invalid tool object or missing input schema\"\n raise ValueError(msg)\n\n input_schema = tool_obj.args_schema\n if not input_schema:\n msg = f\"Empty input schema for tool '{tool_obj.name}'\"\n raise ValueError(msg)\n\n schema_inputs = schema_to_langflow_inputs(input_schema)\n if not schema_inputs:\n msg = f\"No input parameters defined for tool '{tool_obj.name}'\"\n await logger.awarning(msg)\n return []\n\n except Exception as e:\n msg = f\"Error validating schema inputs: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return schema_inputs\n\n async def update_tool_list(self, mcp_server_value=None):\n # Accepts mcp_server_value as dict {name, config} or uses self.mcp_server\n mcp_server = mcp_server_value if mcp_server_value is not None else getattr(self, \"mcp_server\", None)\n server_name = None\n server_config_from_value = None\n if isinstance(mcp_server, dict):\n server_name = mcp_server.get(\"name\")\n server_config_from_value = mcp_server.get(\"config\")\n else:\n server_name = mcp_server\n if not server_name:\n self.tools = []\n await logger.adebug(\"MCP update_tool_list: empty server_name, clearing tools\")\n return [], {\"name\": server_name, \"config\": server_config_from_value}\n\n servers_cache_key = self._mcp_servers_cache_key(server_name)\n\n # Check if caching is enabled, default to False\n use_cache = getattr(self, \"use_cache\", False)\n header_keys = sorted(self._normalized_headers_for_cache().keys())\n await logger.adebug(\n \"MCP update_tool_list: start server=%r use_cache=%s shared_cache_key=%r header_keys=%s\",\n server_name,\n use_cache,\n servers_cache_key,\n header_keys,\n )\n\n async with self._update_tool_list_lock:\n # Use shared cache if available and caching is enabled\n cached = None\n if use_cache:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n cached = servers_cache.get(servers_cache_key) if isinstance(servers_cache, dict) else None\n\n if cached is not None:\n try:\n tools_from_cache = cached[\"tools\"]\n server_config_from_value = cached[\"config\"]\n except (TypeError, KeyError, AttributeError) as e:\n # Handle corrupted cache data by clearing it and continuing to fetch fresh tools\n msg = f\"Unable to use cached data for MCP Server{server_name}: {e}\"\n await logger.awarning(msg)\n # Clear the corrupted cache entry\n current_servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(current_servers_cache, dict) and servers_cache_key in current_servers_cache:\n current_servers_cache.pop(servers_cache_key)\n safe_cache_set(self._shared_component_cache, \"servers\", current_servers_cache)\n else:\n self.tools = tools_from_cache\n self.tool_names = [t.name for t in self.tools if hasattr(t, \"name\")]\n self._tool_cache = cached[\"tool_cache\"]\n await logger.adebug(\n \"MCP update_tool_list: shared_servers_cache HIT count=%d server=%r\",\n len(self.tools),\n server_name,\n )\n return self.tools, {\"name\": server_name, \"config\": server_config_from_value}\n\n try:\n # Try to fetch from database first to ensure we have the latest config.\n # This ensures database updates (like editing a server) take effect.\n # When running in LFX standalone mode the full Langflow package and\n # database may not be available — in that case we skip the DB lookup\n # and fall back to the config embedded in the flow (server_config_from_value).\n server_config_from_db = None\n try:\n from langflow.api.v2.mcp import get_server\n from langflow.services.database.models.user.crud import get_user_by_id\n\n from lfx.services.deps import get_settings_service\n except ModuleNotFoundError as e:\n # Deliberately `except ModuleNotFoundError` (not `except ImportError`): a\n # plain ImportError here means `get_server` / `get_user_by_id` was removed\n # from an installed Langflow — a real API break that should NOT be\n # swallowed as \"standalone mode\". ModuleNotFoundError alone covers the\n # \"Langflow absent\" case.\n #\n # Even within ModuleNotFoundError, only treat this as LFX standalone mode\n # when one of the target Langflow modules is itself missing. Transitive\n # ModuleNotFoundError (e.g. a dependency like sqlmodel failing to import\n # inside langflow.*) indicates a real bug in the full Langflow stack and\n # must surface — otherwise we would silently use a stale flow-embedded\n # config when DB config should have taken precedence.\n missing_module = e.name or \"\"\n is_langflow_standalone = missing_module == \"langflow\" or missing_module.startswith(\"langflow.\")\n if not is_langflow_standalone:\n raise\n await logger.ainfo(\n \"Langflow package not available; using MCP server config from flow value (LFX standalone mode).\"\n )\n else:\n async with session_scope() as db:\n if not self.user_id:\n msg = \"User ID is required for fetching MCP tools.\"\n raise ValueError(msg)\n current_user = await get_user_by_id(db, self.user_id)\n\n # Try to get server config from DB/API\n server_config_from_db = await get_server(\n server_name,\n current_user,\n db,\n storage_service=get_storage_service(),\n settings_service=get_settings_service(),\n )\n\n # Resolve config with proper precedence: DB takes priority, falls back to value\n server_config = resolve_mcp_config(\n server_name=server_name,\n server_config_from_value=server_config_from_value,\n server_config_from_db=server_config_from_db,\n )\n\n if not server_config:\n self.tools = []\n await logger.awarning(\n \"MCP update_tool_list: no server_config after resolve server=%r\",\n server_name,\n )\n return [], {\"name\": server_name, \"config\": server_config}\n\n # Add verify_ssl option to server config if not present\n if \"verify_ssl\" not in server_config:\n verify_ssl = getattr(self, \"verify_ssl\", True)\n server_config[\"verify_ssl\"] = verify_ssl\n\n # Merge headers from component input with server config headers\n # Component headers take precedence over server config headers\n component_headers = getattr(self, \"headers\", None) or []\n if component_headers:\n # Convert list of {\"key\": k, \"value\": v} to dict\n component_headers_dict = {}\n if isinstance(component_headers, list):\n for item in component_headers:\n if isinstance(item, dict) and \"key\" in item and \"value\" in item:\n component_headers_dict[item[\"key\"]] = item[\"value\"]\n elif isinstance(component_headers, dict):\n component_headers_dict = component_headers\n\n if component_headers_dict:\n existing_headers = server_config.get(\"headers\", {}) or {}\n # Ensure existing_headers is a dict (convert from list if needed)\n if isinstance(existing_headers, list):\n existing_dict = {}\n for item in existing_headers:\n if isinstance(item, dict) and \"key\" in item and \"value\" in item:\n existing_dict[item[\"key\"]] = item[\"value\"]\n existing_headers = existing_dict\n merged_headers = {**existing_headers, **component_headers_dict}\n server_config[\"headers\"] = merged_headers\n # Get request_variables from graph context for global variable resolution\n request_variables = None\n if hasattr(self, \"graph\") and self.graph and hasattr(self.graph, \"context\"):\n request_variables = self.graph.context.get(\"request_variables\")\n\n # Only load global variables from database if we have headers that might use them\n # This avoids unnecessary database queries when headers are empty\n has_headers = server_config.get(\"headers\") and len(server_config.get(\"headers\", {})) > 0\n if not request_variables and has_headers:\n try:\n from lfx.services.deps import get_variable_service\n\n variable_service = get_variable_service()\n if variable_service:\n async with session_scope() as db:\n request_variables = await variable_service.get_all_decrypted_variables(\n user_id=self.user_id, session=db\n )\n except Exception as e: # noqa: BLE001\n await logger.awarning(f\"Failed to load global variables for MCP component: {e}\")\n\n await logger.adebug(\n \"MCP update_tool_list: calling update_tools server=%r mode_headers=%s\",\n server_name,\n sorted((server_config.get(\"headers\") or {}).keys())\n if isinstance(server_config.get(\"headers\"), dict)\n else \"list-or-empty\",\n )\n\n _, tool_list, tool_cache = await update_tools(\n server_name=server_name,\n server_config=server_config,\n mcp_stdio_client=self.stdio_client,\n mcp_streamable_http_client=self.streamable_http_client,\n request_variables=request_variables,\n )\n\n self.tool_names = [tool.name for tool in tool_list if hasattr(tool, \"name\")]\n self._tool_cache = tool_cache\n self.tools = tool_list\n\n await logger.adebug(\n \"MCP update_tool_list: fetched from MCP count=%d server=%r\",\n len(tool_list),\n server_name,\n )\n\n # Cache the result only if caching is enabled\n if use_cache:\n cache_data = {\n \"tools\": tool_list,\n \"tool_names\": self.tool_names,\n \"tool_cache\": tool_cache,\n \"config\": server_config,\n }\n\n # Safely update the servers cache with bounded size (FIFO eviction).\n current_servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(current_servers_cache, dict):\n # Because the cache key now includes a header hash, a tenant that\n # rotates session tokens would grow this map without bound. Drop\n # the oldest entry when over the limit.\n max_entries = self.SHARED_SERVERS_CACHE_MAX_ENTRIES\n while (\n len(current_servers_cache) >= max_entries and servers_cache_key not in current_servers_cache\n ):\n oldest_key = next(iter(current_servers_cache))\n current_servers_cache.pop(oldest_key, None)\n current_servers_cache[servers_cache_key] = cache_data\n safe_cache_set(self._shared_component_cache, \"servers\", current_servers_cache)\n await logger.adebug(\n \"MCP update_tool_list: wrote shared_servers_cache key=%r size=%d\",\n servers_cache_key,\n len(current_servers_cache),\n )\n\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = (\n f\"Timeout updating tool list: {e!s}. \"\n \"Raise ``LANGFLOW_MCP_SERVER_TIMEOUT`` for the deployment if the MCP \"\n \"server legitimately needs more time to respond.\"\n )\n await logger.aexception(msg)\n raise TimeoutError(msg) from e\n except Exception as e:\n msg = f\"Error updating tool list: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return tool_list, {\"name\": server_name, \"config\": server_config}\n\n async def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None) -> dict:\n \"\"\"Toggle the visibility of connection-specific fields based on the selected mode.\"\"\"\n try:\n is_refresh = bool(build_config.get(\"is_refresh\", False))\n if field_name == \"tool\":\n try:\n # Always refresh tools when cache is disabled, or when tools list is empty\n # This ensures database edits are reflected immediately when cache is disabled\n use_cache = getattr(self, \"use_cache\", False)\n if is_refresh:\n self.tools = []\n if is_refresh or len(self.tools) == 0 or not use_cache:\n try:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n build_config[\"tool\"][\"options\"] = [tool.name for tool in self.tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout updating tool list: {e!s}\"\n await logger.aexception(msg)\n if not build_config[\"tools_metadata\"][\"show\"]:\n build_config[\"tool\"][\"show\"] = True\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = msg\n else:\n build_config[\"tool\"][\"show\"] = False\n except ValueError as e:\n msg = f\"Error updating tool list: {e!s}\"\n await logger.aexception(msg)\n if not build_config[\"tools_metadata\"][\"show\"]:\n build_config[\"tool\"][\"show\"] = True\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = msg\n else:\n build_config[\"tool\"][\"show\"] = False\n\n if field_value == \"\":\n return build_config\n tool_obj = None\n for tool in self.tools:\n if tool.name == field_value:\n tool_obj = tool\n break\n if tool_obj is None:\n msg = f\"Tool {field_value} not found in available tools: {self.tools}\"\n await logger.awarning(msg)\n return build_config\n await self._update_tool_config(build_config, field_value)\n except Exception as e:\n build_config[\"tool\"][\"options\"] = []\n msg = f\"Failed to update tools: {e!s}\"\n raise ValueError(msg) from e\n else:\n return build_config\n elif field_name == \"mcp_server\":\n if not field_value:\n build_config[\"tool\"][\"show\"] = False\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = \"\"\n build_config[\"tool_placeholder\"][\"tool_mode\"] = False\n self.remove_non_default_keys(build_config)\n return build_config\n\n build_config[\"tool_placeholder\"][\"tool_mode\"] = True\n\n current_server_name = field_value.get(\"name\") if isinstance(field_value, dict) else field_value\n build_config_server_value = build_config.get(\"mcp_server\", {}).get(\"value\")\n build_config_server_name = (\n build_config_server_value.get(\"name\")\n if isinstance(build_config_server_value, dict)\n else build_config_server_value\n )\n servers_cache_key_ui = self._mcp_servers_cache_key(current_server_name) if current_server_name else \"\"\n _last_selected_server = safe_cache_get(self._shared_component_cache, \"last_selected_server\", \"\")\n # Only treat as a server change if there was a previous server selection.\n # Cold cache (_last_selected_server=\"\") on initial flow load is NOT a server change —\n # the user didn't switch anything, the backend just hasn't seen this component yet.\n server_changed = bool(\n (_last_selected_server and current_server_name != _last_selected_server)\n or (build_config_server_name and current_server_name != build_config_server_name)\n )\n\n # Determine if \"Tool Mode\" is active by checking if the tool dropdown is hidden.\n is_in_tool_mode = build_config[\"tools_metadata\"][\"show\"]\n\n # Get use_cache setting to determine if we should use cached data\n use_cache = getattr(self, \"use_cache\", False)\n\n # Fast path: if server didn't change and we already have options, keep them as-is\n # BUT only if caching is enabled, we're in tool mode, or it's the initial load\n existing_options = build_config.get(\"tool\", {}).get(\"options\") or []\n if not is_refresh and not server_changed and existing_options:\n # In non-tool mode with cache disabled, skip the fast path to force refresh\n # BUT on initial load (cold cache), always preserve saved options from the flow\n if not is_in_tool_mode and not use_cache and _last_selected_server:\n pass # Continue to refresh logic below (user-initiated with cache disabled)\n else:\n if not is_in_tool_mode:\n build_config[\"tool\"][\"show\"] = True\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", current_server_name)\n return build_config\n\n # To avoid unnecessary updates, only proceed if the server has actually changed\n # OR if caching is disabled (to force refresh in non-tool mode)\n if (\n not is_refresh\n and (_last_selected_server in (current_server_name, \"\"))\n and build_config[\"tool\"][\"show\"]\n and use_cache\n ):\n if current_server_name:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict):\n cached = servers_cache.get(servers_cache_key_ui)\n if cached is not None and cached.get(\"tool_names\"):\n cached_tools = cached[\"tool_names\"]\n current_tools = build_config[\"tool\"][\"options\"]\n if current_tools == cached_tools:\n return build_config\n else:\n return build_config\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", current_server_name)\n\n # When cache is disabled, clear any cached data for this server\n # This ensures we always fetch fresh data from the database\n if (is_refresh or not use_cache) and current_server_name:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict) and servers_cache_key_ui in servers_cache:\n servers_cache.pop(servers_cache_key_ui)\n safe_cache_set(self._shared_component_cache, \"servers\", servers_cache)\n\n # Check if tools are already cached for this server before clearing\n cached_tools = None\n if current_server_name and use_cache and not is_refresh:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict):\n cached = servers_cache.get(servers_cache_key_ui)\n if cached is not None:\n try:\n cached_tools = cached[\"tools\"]\n self.tools = cached_tools\n self.tool_names = cached[\"tool_names\"]\n self._tool_cache = cached[\"tool_cache\"]\n except (TypeError, KeyError, AttributeError) as e:\n # Handle corrupted cache data by ignoring it\n msg = f\"Unable to use cached data for MCP Server,{current_server_name}: {e}\"\n await logger.awarning(msg)\n cached_tools = None\n\n # Clear tools when cache is disabled OR when we don't have cached tools\n # This ensures fresh tools are fetched after database edits\n if is_refresh or not cached_tools or not use_cache:\n self.tools = [] # Clear previous tools to force refresh\n\n # Clear previous tool inputs if:\n # 1. Server actually changed\n # 2. Cache is disabled (meaning tool list will be refreshed)\n if is_refresh or server_changed or not use_cache:\n self.remove_non_default_keys(build_config)\n\n # Only show the tool dropdown if not in tool_mode\n if not is_in_tool_mode:\n build_config[\"tool\"][\"show\"] = True\n if cached_tools:\n # Use cached tools to populate options immediately\n build_config[\"tool\"][\"options\"] = [tool.name for tool in cached_tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n else:\n # Actually fetch tools now instead of deferring to a frontend callback.\n # The frontend has no reliable mechanism to trigger a second\n # update_build_config call for the \"tool\" field after this response,\n # so we must populate the options here.\n try:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list(\n mcp_server_value=field_value\n )\n build_config[\"tool\"][\"options\"] = [tool.name for tool in self.tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout loading tools for MCP server: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = \"Timeout on MCP server\"\n except (ValueError, ImportError, ConnectionError, OSError, RuntimeError) as e:\n msg = f\"Error loading tools for MCP server: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = (\n \"Error on MCP Server\" if \"'NoneType' object has no attribute 'id'\" in msg else msg\n )\n # Force a value refresh only when the user genuinely switched servers.\n # server_changed is only True for real user-initiated changes (not initial load).\n if server_changed:\n build_config[\"tool\"][\"value\"] = uuid.uuid4()\n else:\n # Keep the tool dropdown hidden if in tool_mode\n self._not_load_actions = True\n build_config[\"tool\"][\"show\"] = False\n\n elif field_name == \"tool_mode\":\n build_config[\"tool\"][\"placeholder\"] = \"\"\n build_config[\"tool\"][\"show\"] = not bool(field_value) and bool(build_config[\"mcp_server\"])\n self.remove_non_default_keys(build_config)\n self.tool = build_config[\"tool\"][\"value\"]\n if field_value:\n self._not_load_actions = True\n else:\n build_config[\"tool\"][\"value\"] = uuid.uuid4()\n build_config[\"tool\"][\"show\"] = True\n # Fetch tools immediately instead of showing \"Loading tools...\"\n try:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n build_config[\"tool\"][\"options\"] = [tool.name for tool in self.tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout loading tools when toggling tool mode: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = msg\n except (ValueError, ImportError, ConnectionError, OSError, RuntimeError) as e:\n msg = f\"Error loading tools when toggling tool mode: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = msg\n elif field_name == \"tools_metadata\":\n self._not_load_actions = False\n\n except Exception as e:\n msg = f\"Error in update_build_config: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return build_config\n\n @staticmethod\n def _unwrap_optional_annotation(annotation: Any) -> Any:\n \"\"\"Remove a single None branch from a union annotation.\"\"\"\n if isinstance(annotation, UnionType):\n non_none = [item for item in get_args(annotation) if item is not type(None)]\n if len(non_none) == 1:\n return non_none[0]\n return annotation\n\n if get_origin(annotation) is None:\n return annotation\n\n non_none = [item for item in get_args(annotation) if item is not type(None)]\n if len(non_none) == 1 and len(non_none) != len(get_args(annotation)):\n return non_none[0]\n return annotation\n\n @classmethod\n def _is_object_like_annotation(cls, annotation: Any) -> bool:\n \"\"\"Return True when the annotation represents a dict-like payload.\"\"\"\n annotation = cls._unwrap_optional_annotation(annotation)\n origin = get_origin(annotation)\n if origin is dict:\n return True\n return annotation is dict or (isinstance(annotation, type) and issubclass(annotation, BaseModel))\n\n @classmethod\n def _should_include_tool_argument(cls, model_field: Any, value: Any) -> bool:\n \"\"\"Omit blank optional values so MCP server defaults remain intact.\"\"\"\n if value is None:\n return False\n\n if model_field.is_required():\n return True\n\n if isinstance(value, str) and value == \"\":\n return False\n\n return not (\n value == {} and model_field.default is None and cls._is_object_like_annotation(model_field.annotation)\n )\n\n def _build_tool_kwargs(self, args_schema: type[BaseModel]) -> dict[str, Any]:\n \"\"\"Collect tool kwargs from component inputs, omitting blank optional values.\"\"\"\n kwargs: dict[str, Any] = {}\n for arg_name, model_field in args_schema.model_fields.items():\n value = getattr(self, arg_name, None)\n if isinstance(value, Message):\n value = value.text\n\n if self._should_include_tool_argument(model_field, value):\n kwargs[arg_name] = value\n\n return kwargs\n\n def get_inputs_for_all_tools(self, tools: list) -> dict:\n \"\"\"Get input schemas for all tools.\"\"\"\n inputs = {}\n for tool in tools:\n if not tool or not hasattr(tool, \"name\"):\n continue\n try:\n langflow_inputs = schema_to_langflow_inputs(tool.args_schema)\n inputs[tool.name] = langflow_inputs\n except (AttributeError, ValueError, TypeError, KeyError) as e:\n msg = f\"Error getting inputs for tool {getattr(tool, 'name', 'unknown')}: {e!s}\"\n logger.exception(msg)\n continue\n return inputs\n\n def remove_non_default_keys(self, build_config: dict) -> None:\n \"\"\"Remove non-default keys from the build config.\"\"\"\n for key in list(build_config.keys()):\n if key not in self.default_keys:\n build_config.pop(key)\n\n async def _update_tool_config(self, build_config: dict, tool_name: str) -> None:\n \"\"\"Update tool configuration with proper error handling.\"\"\"\n if not self.tools:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n\n if not tool_name:\n return\n\n tool_obj = next((tool for tool in self.tools if tool.name == tool_name), None)\n if not tool_obj:\n msg = f\"Tool {tool_name} not found in available tools: {self.tools}\"\n self.remove_non_default_keys(build_config)\n build_config[\"tool\"][\"value\"] = \"\"\n await logger.awarning(msg)\n return\n\n try:\n # Store current values before removing inputs (only for the current tool)\n current_values = {}\n for key, value in build_config.items():\n if key not in self.default_keys and isinstance(value, dict) and \"value\" in value:\n current_values[key] = value[\"value\"]\n\n # Remove ALL non-default keys (all previous tool inputs)\n self.remove_non_default_keys(build_config)\n\n # Get and validate new inputs for the selected tool\n self.schema_inputs = await self._validate_schema_inputs(tool_obj)\n if not self.schema_inputs:\n msg = f\"No input parameters to configure for tool '{tool_name}'\"\n await logger.ainfo(msg)\n return\n\n # Add new inputs to build config for the selected tool only\n for schema_input in self.schema_inputs:\n if not schema_input or not hasattr(schema_input, \"name\"):\n msg = \"Invalid schema input detected, skipping\"\n await logger.awarning(msg)\n continue\n\n try:\n name = schema_input.name\n input_dict = schema_input.to_dict()\n input_dict.setdefault(\"value\", None)\n input_dict.setdefault(\"required\", True)\n\n build_config[name] = input_dict\n\n # Preserve existing value if the parameter name exists in current_values\n if name in current_values:\n build_config[name][\"value\"] = current_values[name]\n\n except (AttributeError, KeyError, TypeError) as e:\n msg = f\"Error processing schema input {schema_input}: {e!s}\"\n await logger.aexception(msg)\n continue\n except ValueError as e:\n msg = f\"Schema validation error for tool {tool_name}: {e!s}\"\n await logger.aexception(msg)\n self.schema_inputs = []\n return\n except (AttributeError, KeyError, TypeError) as e:\n msg = f\"Error updating tool config: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n\n async def build_output(self) -> DataFrame:\n \"\"\"Build output with improved error handling and validation.\"\"\"\n try:\n self.tools, _ = await self.update_tool_list()\n if self.tool != \"\":\n # Set session context for persistent MCP sessions using Langflow session ID\n session_context = self._get_session_context()\n if session_context:\n self.stdio_client.set_session_context(session_context)\n self.streamable_http_client.set_session_context(session_context)\n exec_tool = self._tool_cache[self.tool]\n kwargs = self._build_tool_kwargs(exec_tool.args_schema)\n unflattened_kwargs = maybe_unflatten_dict(kwargs)\n\n output = await exec_tool.coroutine(**unflattened_kwargs)\n tool_content = []\n for item in output.content:\n item_dict = item.model_dump()\n item_dict = self.process_output_item(item_dict)\n tool_content.append(item_dict)\n\n if isinstance(tool_content, list) and all(isinstance(x, dict) for x in tool_content):\n return DataFrame(tool_content)\n return DataFrame(data=tool_content)\n return DataFrame(data=[{\"error\": \"You must select a tool\"}])\n except Exception as e:\n msg = f\"Error in build_output: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n\n def process_output_item(self, item_dict):\n \"\"\"Process the output of a tool.\"\"\"\n if item_dict.get(\"type\") == \"text\":\n text = item_dict.get(\"text\")\n try:\n parsed = json.loads(text)\n # Ensure we always return a dictionary for DataFrame compatibility\n if isinstance(parsed, dict):\n return parsed\n # Wrap non-dict parsed values in a dictionary\n return {\"text\": text, \"parsed_value\": parsed, \"type\": \"text\"} # noqa: TRY300\n except json.JSONDecodeError:\n return item_dict\n return item_dict\n\n def _get_session_context(self) -> str | None:\n \"\"\"Get the Langflow session ID for MCP session caching.\"\"\"\n # Try to get session ID from the component's execution context\n if hasattr(self, \"graph\") and hasattr(self.graph, \"session_id\"):\n session_id = self.graph.session_id\n # Include server name to ensure different servers get different sessions\n server_name = \"\"\n mcp_server = getattr(self, \"mcp_server\", None)\n if isinstance(mcp_server, dict):\n server_name = mcp_server.get(\"name\", \"\")\n elif mcp_server:\n server_name = str(mcp_server)\n return f\"{session_id}_{server_name}\" if session_id else None\n return None\n\n async def _get_tools(self):\n \"\"\"Load tools for the agent toolkit; always refresh from ``update_tool_list``.\n\n ``_not_load_actions`` only applies to UI build_config flows (tool dropdown). Agent\n runs must always bind the current tool list (including after header/tweak changes).\n\n A short-lived TTL cache (``TOOL_TTL_SECS``, header-hash-keyed) skips the MCP\n round-trip when the same auth context is re-queried quickly (e.g. parallel agent\n steps sharing the same tweaked headers). The cache is per component instance and\n bounded by ``TOOL_TTL_MAX_ENTRIES``; it is distinct from the \"Use Cached Server\"\n (``use_cache``) toggle which controls the shared cross-request cache.\n \"\"\"\n mcp_server = getattr(self, \"mcp_server\", None)\n srv = mcp_server.get(\"name\") if isinstance(mcp_server, dict) else mcp_server\n\n ttl = self.TOOL_TTL_SECS\n ttl_key = self._mcp_servers_cache_key(srv or \"\") if ttl > 0 and srv else \"\"\n\n # TTL cache lookup + expired-entry eviction.\n if ttl > 0 and ttl_key:\n cached = self._ttl_tool_cache.get(ttl_key)\n if cached is not None:\n ts, cached_tools = cached\n age = time.monotonic() - ts\n if age < ttl:\n await logger.adebug(\n \"MCP _get_tools: TTL cache hit count=%d age=%.1fs server=%r\",\n len(cached_tools),\n age,\n srv,\n )\n return cached_tools\n # Stale — drop it so the size check below stays tight.\n self._ttl_tool_cache.pop(ttl_key, None)\n\n await logger.adebug(\"MCP _get_tools: fetching tool list server=%r\", srv)\n tools, _ = await self.update_tool_list(mcp_server)\n await logger.adebug(\"MCP _get_tools: fetched %d tools server=%r\", len(tools), srv)\n\n # TTL cache store with bounded size (FIFO eviction of the oldest entry).\n if ttl > 0 and ttl_key and tools:\n if len(self._ttl_tool_cache) >= self.TOOL_TTL_MAX_ENTRIES:\n # dict preserves insertion order — pop the oldest entry.\n oldest_key = next(iter(self._ttl_tool_cache))\n self._ttl_tool_cache.pop(oldest_key, None)\n self._ttl_tool_cache[ttl_key] = (time.monotonic(), tools)\n\n return tools\n" + "value": "from __future__ import annotations\n\nimport asyncio\nimport hashlib\nimport json\nimport time\nimport uuid\nfrom types import UnionType\nfrom typing import Any, get_args, get_origin\n\nfrom langchain_core.tools import StructuredTool # noqa: TC002\nfrom pydantic import BaseModel\n\nfrom lfx.base.agents.utils import maybe_unflatten_dict, safe_cache_get, safe_cache_set\nfrom lfx.base.mcp.util import (\n MCPStdioClient,\n MCPStreamableHttpClient,\n update_tools,\n)\nfrom lfx.base.tools.constants import TOOL_OUTPUT_DISPLAY_NAME, TOOL_OUTPUT_NAME\nfrom lfx.custom.custom_component.component_with_cache import ComponentWithCache\nfrom lfx.inputs.inputs import InputTypes # noqa: TC001\nfrom lfx.io import BoolInput, DictInput, DropdownInput, FloatInput, McpInput, MessageTextInput, Output\nfrom lfx.io.schema import schema_to_langflow_inputs\nfrom lfx.log.logger import logger\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.services.deps import get_storage_service, session_scope\n\n\ndef resolve_mcp_config(\n server_name: str, # noqa: ARG001\n server_config_from_value: dict | None,\n server_config_from_db: dict | None,\n) -> dict | None:\n \"\"\"Resolve MCP server config with proper precedence.\n\n Resolves the configuration for an MCP server with the following precedence:\n 1. Database config (takes priority) - ensures edits are reflected\n 2. Config from value/tweaks (fallback) - allows REST API to provide config for new servers\n\n Args:\n server_name: Name of the MCP server\n server_config_from_value: Config provided via value/tweaks (optional)\n server_config_from_db: Config from database (optional)\n\n Returns:\n Final config to use (DB takes priority, falls back to value)\n Returns None if no config found in either location\n \"\"\"\n if server_config_from_db:\n return server_config_from_db\n return server_config_from_value\n\n\n# TODO(legacy-cleanup): This file is ~800 lines, over the 500-line guideline. Split\n# helper functions (resolve_mcp_config, connection resolution) from the MCPToolsComponent\n# orchestration logic in a follow-up PR.\nclass MCPToolsComponent(ComponentWithCache):\n \"\"\"MCP Tools component.\n\n Behaviour notes:\n - Stale agent tools vs server were caused by Langflow caching the Toolset output\n (``Output.cache=True`` plus the persisted ``output.value`` in saved flow JSON), not\n only by the user-facing \"Use Cached Server\" (``use_cache``) toggle. ``_build_tool_output``\n declares the Toolset output with ``cache=False`` and ``map_outputs`` overrides the\n persisted value for existing flows so every run resolves a fresh tool list.\n - ``update_tool_list`` is serialized with an ``asyncio.Lock`` so concurrent calls do not\n share the same Streamable HTTP client session; overlapping POST/DELETE cycles otherwise\n surface as HTTP 404 and the MCP SDK reports ``Session terminated``.\n - All diagnostic logs are at DEBUG level to avoid swamping logs on hot paths (an agent\n may call ``_get_tools`` per step). Header values are never logged; only header *keys*.\n \"\"\"\n\n # Short-lived in-memory cache window (seconds) for ``_get_tools``. When the\n # same header-hash key is requested again within this window the cached tool\n # list is reused so parallel agent runs that share identical auth don't\n # each pay for a fresh MCP round-trip. Set to ``0`` to disable. This is\n # distinct from the \"Use Cached Server\" (``use_cache``) toggle which only\n # controls the shared cross-request server cache.\n TOOL_TTL_SECS: int = 30\n # Upper bound on the per-instance TTL cache. Oldest entries are evicted\n # when the cap is reached. Keeps memory bounded for flows where the same\n # component handles many rotating auth contexts (e.g. per-tenant tokens).\n TOOL_TTL_MAX_ENTRIES: int = 32\n # Upper bound on the shared cross-request ``servers`` cache. Each\n # (server_name, header-hash) pair is a distinct entry; without a bound a\n # tenant that rotates session tokens would grow this map without limit.\n SHARED_SERVERS_CACHE_MAX_ENTRIES: int = 64\n\n schema_inputs: list = []\n tools: list[StructuredTool] = []\n _not_load_actions: bool = False\n _tool_cache: dict = {}\n _last_selected_server: str | None = None # Cache for the last selected server\n\n def __init__(self, **data) -> None:\n super().__init__(**data)\n # Initialize cache keys to avoid CacheMiss when accessing them\n self._ensure_cache_structure()\n\n # Initialize clients with access to the component cache.\n # Per-component timeout is normalized and applied immediately before MCP calls.\n self.stdio_client: MCPStdioClient = MCPStdioClient(component_cache=self._shared_component_cache)\n self.streamable_http_client: MCPStreamableHttpClient = MCPStreamableHttpClient(\n component_cache=self._shared_component_cache\n )\n # One MCP stdio/streamable client pair per component; concurrent update_tool_list calls\n # otherwise race (session DELETE vs POST) and the MCP SDK surfaces HTTP 404 as \"Session terminated\".\n self._update_tool_list_lock = asyncio.Lock()\n # Per-instance TTL cache for ``_get_tools``: {cache_key: (monotonic_ts, tools)}.\n # Declared here (not at class scope) so every component gets its own dict —\n # a class-level dict would be shared across every MCPToolsComponent in the process,\n # leaking tool lists across tenants that happen to hash to the same key.\n self._ttl_tool_cache: dict[str, tuple[float, list]] = {}\n\n def _ensure_cache_structure(self):\n \"\"\"Ensure the cache has the required structure.\"\"\"\n # Check if servers key exists and is not CacheMiss\n servers_value = safe_cache_get(self._shared_component_cache, \"servers\")\n if servers_value is None:\n safe_cache_set(self._shared_component_cache, \"servers\", {})\n\n # Check if last_selected_server key exists and is not CacheMiss\n last_server_value = safe_cache_get(self._shared_component_cache, \"last_selected_server\")\n if last_server_value is None:\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", \"\")\n\n def _normalized_headers_for_cache(self) -> dict[str, str]:\n \"\"\"Component headers as a dict for stable cache keying (auth / tweaks).\"\"\"\n component_headers = getattr(self, \"headers\", None) or []\n if isinstance(component_headers, list):\n return {\n str(item[\"key\"]): str(item[\"value\"])\n for item in component_headers\n if isinstance(item, dict) and \"key\" in item and \"value\" in item\n }\n if isinstance(component_headers, dict):\n return {str(k): str(v) for k, v in component_headers.items()}\n return {}\n\n def _normalize_tool_execution_timeout(self) -> float | None:\n \"\"\"Normalize the timeout input and reject negative values with a field-specific error.\"\"\"\n timeout_value = getattr(self, \"tool_execution_timeout\", 0.0)\n\n if timeout_value in (None, \"\"):\n return None\n\n try:\n val = float(timeout_value)\n except (ValueError, TypeError):\n return None\n\n if val < 0:\n msg = \"Tool Execution Timeout must be greater than or equal to 0.\"\n raise ValueError(msg)\n\n return val if val else None\n\n def _mcp_servers_cache_key(self, server_name: str) -> str:\n \"\"\"Cache key for shared servers map.\n\n Includes headers and timeout so auth/tweak/timeout changes get distinct entries.\n \"\"\"\n if not server_name:\n return \"\"\n\n raw_timeout = getattr(self, \"tool_execution_timeout\", 0.0) or 0.0\n normalized_timeout = max(0.0, float(raw_timeout))\n\n hdrs = self._normalized_headers_for_cache()\n\n # Build cache key components\n cache_data = {\n \"headers\": hdrs,\n \"timeout\": normalized_timeout,\n }\n\n # If no headers and default timeout, just use server name\n if not hdrs and normalized_timeout == 0.0:\n return server_name\n\n payload = json.dumps(cache_data, sort_keys=True)\n digest = hashlib.sha256(payload.encode()).hexdigest()[:16]\n return f\"{server_name}:{digest}\"\n\n def _build_tool_output(self) -> Output:\n # Do not cache Toolset output. This is separate from the MCP \"Use Cached Server\" (use_cache)\n # toggle: Langflow's Output.cache defaults to True and was memoizing the first to_toolkit()\n # result, so per-request tweaks/headers never refreshed bound tools even when use_cache=False.\n return Output(\n name=TOOL_OUTPUT_NAME,\n display_name=TOOL_OUTPUT_DISPLAY_NAME,\n method=\"to_toolkit\",\n types=[\"Tool\"],\n cache=False,\n )\n\n def map_outputs(self) -> None:\n \"\"\"Override the persisted ``component_as_tool`` cache flag from saved flow JSON.\n\n ``_build_tool_output`` already returns the output with ``cache=False``, but the\n flow JSON for existing flows often stores ``cache: true`` for ``component_as_tool``\n and that persisted value wins over the declaration. Forcing ``cache=False`` here\n guarantees saved flows also bypass Output memoization and get a fresh tool list\n on every run.\n \"\"\"\n super().map_outputs()\n if TOOL_OUTPUT_NAME in self._outputs_map:\n self._outputs_map[TOOL_OUTPUT_NAME].cache = False\n\n default_keys: list[str] = [\n \"code\",\n \"_type\",\n \"tool_mode\",\n \"tool_placeholder\",\n \"mcp_server\",\n \"tool\",\n \"use_cache\",\n \"verify_ssl\",\n \"headers\",\n \"tool_execution_timeout\",\n ]\n\n display_name = \"MCP Tools\"\n description = \"Connect to an MCP server to use its tools.\"\n documentation: str = \"https://docs.langflow.org/mcp-tools\"\n icon = \"Mcp\"\n name = \"MCPTools\"\n\n inputs = [\n McpInput(\n name=\"mcp_server\",\n display_name=\"MCP Server\",\n info=\"Select the MCP Server that will be used by this component\",\n real_time_refresh=True,\n ),\n BoolInput(\n name=\"use_cache\",\n display_name=\"Use Cached Server\",\n info=(\n \"Enable caching of MCP Server and tools to improve performance. \"\n \"Disable to always fetch fresh tools and server updates.\"\n ),\n value=False,\n advanced=True,\n ),\n BoolInput(\n name=\"verify_ssl\",\n display_name=\"Verify SSL Certificate\",\n info=(\n \"Enable SSL certificate verification for HTTPS connections. \"\n \"Disable only for development/testing with self-signed certificates.\"\n ),\n value=True,\n advanced=True,\n ),\n DictInput(\n name=\"headers\",\n display_name=\"Headers\",\n info=(\n \"HTTP headers to include with MCP server requests. \"\n \"Useful for authentication (e.g., Authorization header). \"\n \"These headers override any headers configured in the MCP server settings.\"\n ),\n advanced=True,\n is_list=True,\n ),\n FloatInput(\n name=\"tool_execution_timeout\",\n display_name=\"Tool Execution Timeout (seconds)\",\n info=(\n \"Maximum time to wait for tool execution before timing out. \"\n \"Supports decimal values for sub-second timeouts (e.g., 0.01 for 10ms). \"\n \"Set to 0 to use the system-configured MCP timeout.\"\n ),\n value=0.0,\n range_spec={\"min\": 0.0, \"max\": 3600.0, \"step\": 0.01},\n real_time_refresh=True,\n advanced=True,\n ),\n DropdownInput(\n name=\"tool\",\n display_name=\"Tool\",\n options=[],\n value=\"\",\n info=\"Select the tool to execute\",\n show=False,\n required=True,\n real_time_refresh=True,\n refresh_button=True,\n ),\n MessageTextInput(\n name=\"tool_placeholder\",\n display_name=\"Tool Placeholder\",\n info=\"Placeholder for the tool\",\n value=\"\",\n show=False,\n tool_mode=False,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Response\", name=\"response\", method=\"build_output\"),\n ]\n\n async def _validate_schema_inputs(self, tool_obj) -> list[InputTypes]:\n \"\"\"Validate and process schema inputs for a tool.\"\"\"\n try:\n if not tool_obj or not hasattr(tool_obj, \"args_schema\"):\n msg = \"Invalid tool object or missing input schema\"\n raise ValueError(msg)\n\n input_schema = tool_obj.args_schema\n if not input_schema:\n msg = f\"Empty input schema for tool '{tool_obj.name}'\"\n raise ValueError(msg)\n\n schema_inputs = schema_to_langflow_inputs(input_schema)\n if not schema_inputs:\n msg = f\"No input parameters defined for tool '{tool_obj.name}'\"\n await logger.awarning(msg)\n return []\n\n except Exception as e:\n msg = f\"Error validating schema inputs: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return schema_inputs\n\n async def update_tool_list(self, mcp_server_value=None):\n # Accepts mcp_server_value as dict {name, config} or uses self.mcp_server\n mcp_server = mcp_server_value if mcp_server_value is not None else getattr(self, \"mcp_server\", None)\n server_name = None\n server_config_from_value = None\n if isinstance(mcp_server, dict):\n server_name = mcp_server.get(\"name\")\n server_config_from_value = mcp_server.get(\"config\")\n else:\n server_name = mcp_server\n if not server_name:\n self.tools = []\n await logger.adebug(\"MCP update_tool_list: empty server_name, clearing tools\")\n return [], {\"name\": server_name, \"config\": server_config_from_value}\n\n servers_cache_key = self._mcp_servers_cache_key(server_name)\n\n # Check if caching is enabled, default to False\n use_cache = getattr(self, \"use_cache\", False)\n header_keys = sorted(self._normalized_headers_for_cache().keys())\n await logger.adebug(\n \"MCP update_tool_list: start server=%r use_cache=%s shared_cache_key=%r header_keys=%s\",\n server_name,\n use_cache,\n servers_cache_key,\n header_keys,\n )\n\n async with self._update_tool_list_lock:\n # Use shared cache if available and caching is enabled\n cached = None\n if use_cache:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n cached = servers_cache.get(servers_cache_key) if isinstance(servers_cache, dict) else None\n\n if cached is not None:\n try:\n tools_from_cache = cached[\"tools\"]\n server_config_from_value = cached[\"config\"]\n except (TypeError, KeyError, AttributeError) as e:\n # Handle corrupted cache data by clearing it and continuing to fetch fresh tools\n msg = f\"Unable to use cached data for MCP Server{server_name}: {e}\"\n await logger.awarning(msg)\n # Clear the corrupted cache entry\n current_servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(current_servers_cache, dict) and servers_cache_key in current_servers_cache:\n current_servers_cache.pop(servers_cache_key)\n safe_cache_set(self._shared_component_cache, \"servers\", current_servers_cache)\n else:\n self.tools = tools_from_cache\n self.tool_names = [t.name for t in self.tools if hasattr(t, \"name\")]\n self._tool_cache = cached[\"tool_cache\"]\n await logger.adebug(\n \"MCP update_tool_list: shared_servers_cache HIT count=%d server=%r\",\n len(self.tools),\n server_name,\n )\n return self.tools, {\"name\": server_name, \"config\": server_config_from_value}\n\n try:\n # Try to fetch from database first to ensure we have the latest config.\n # This ensures database updates (like editing a server) take effect.\n # When running in LFX standalone mode the full Langflow package and\n # database may not be available — in that case we skip the DB lookup\n # and fall back to the config embedded in the flow (server_config_from_value).\n server_config_from_db = None\n try:\n from langflow.api.v2.mcp import get_server\n from langflow.services.database.models.user.crud import get_user_by_id\n\n from lfx.services.deps import get_settings_service\n except ModuleNotFoundError as e:\n # Deliberately `except ModuleNotFoundError` (not `except ImportError`): a\n # plain ImportError here means `get_server` / `get_user_by_id` was removed\n # from an installed Langflow — a real API break that should NOT be\n # swallowed as \"standalone mode\". ModuleNotFoundError alone covers the\n # \"Langflow absent\" case.\n #\n # Even within ModuleNotFoundError, only treat this as LFX standalone mode\n # when one of the target Langflow modules is itself missing. Transitive\n # ModuleNotFoundError (e.g. a dependency like sqlmodel failing to import\n # inside langflow.*) indicates a real bug in the full Langflow stack and\n # must surface — otherwise we would silently use a stale flow-embedded\n # config when DB config should have taken precedence.\n missing_module = e.name or \"\"\n is_langflow_standalone = missing_module == \"langflow\" or missing_module.startswith(\"langflow.\")\n if not is_langflow_standalone:\n raise\n await logger.ainfo(\n \"Langflow package not available; using MCP server config from flow value (LFX standalone mode).\"\n )\n else:\n async with session_scope() as db:\n if not self.user_id:\n msg = \"User ID is required for fetching MCP tools.\"\n raise ValueError(msg)\n current_user = await get_user_by_id(db, self.user_id)\n\n # Try to get server config from DB/API\n server_config_from_db = await get_server(\n server_name,\n current_user,\n db,\n storage_service=get_storage_service(),\n settings_service=get_settings_service(),\n )\n\n # Resolve config with proper precedence: DB takes priority, falls back to value\n server_config = resolve_mcp_config(\n server_name=server_name,\n server_config_from_value=server_config_from_value,\n server_config_from_db=server_config_from_db,\n )\n\n if not server_config:\n self.tools = []\n await logger.awarning(\n \"MCP update_tool_list: no server_config after resolve server=%r\",\n server_name,\n )\n return [], {\"name\": server_name, \"config\": server_config}\n\n # Add verify_ssl option to server config if not present\n if \"verify_ssl\" not in server_config:\n verify_ssl = getattr(self, \"verify_ssl\", True)\n server_config[\"verify_ssl\"] = verify_ssl\n\n # Merge headers from component input with server config headers\n # Component headers take precedence over server config headers\n component_headers = getattr(self, \"headers\", None) or []\n if component_headers:\n # Convert list of {\"key\": k, \"value\": v} to dict\n component_headers_dict = {}\n if isinstance(component_headers, list):\n for item in component_headers:\n if isinstance(item, dict) and \"key\" in item and \"value\" in item:\n component_headers_dict[item[\"key\"]] = item[\"value\"]\n elif isinstance(component_headers, dict):\n component_headers_dict = component_headers\n\n if component_headers_dict:\n existing_headers = server_config.get(\"headers\", {}) or {}\n # Ensure existing_headers is a dict (convert from list if needed)\n if isinstance(existing_headers, list):\n existing_dict = {}\n for item in existing_headers:\n if isinstance(item, dict) and \"key\" in item and \"value\" in item:\n existing_dict[item[\"key\"]] = item[\"value\"]\n existing_headers = existing_dict\n merged_headers = {**existing_headers, **component_headers_dict}\n server_config[\"headers\"] = merged_headers\n # Get request_variables from graph context for global variable resolution\n request_variables = None\n if hasattr(self, \"graph\") and self.graph and hasattr(self.graph, \"context\"):\n request_variables = self.graph.context.get(\"request_variables\")\n\n # Only load global variables from database if we have headers that might use them\n # This avoids unnecessary database queries when headers are empty\n has_headers = server_config.get(\"headers\") and len(server_config.get(\"headers\", {})) > 0\n if not request_variables and has_headers:\n try:\n from lfx.services.deps import get_variable_service\n\n variable_service = get_variable_service()\n if variable_service:\n async with session_scope() as db:\n request_variables = await variable_service.get_all_decrypted_variables(\n user_id=self.user_id, session=db\n )\n except Exception as e: # noqa: BLE001\n await logger.awarning(f\"Failed to load global variables for MCP component: {e}\")\n\n await logger.adebug(\n \"MCP update_tool_list: calling update_tools server=%r mode_headers=%s\",\n server_name,\n sorted((server_config.get(\"headers\") or {}).keys())\n if isinstance(server_config.get(\"headers\"), dict)\n else \"list-or-empty\",\n )\n\n timeout = self._normalize_tool_execution_timeout()\n\n _, tool_list, tool_cache = await update_tools(\n server_name=server_name,\n server_config=server_config,\n mcp_stdio_client=self.stdio_client,\n mcp_streamable_http_client=self.streamable_http_client,\n request_variables=request_variables,\n tool_execution_timeout=timeout,\n )\n\n self.tool_names = [tool.name for tool in tool_list if hasattr(tool, \"name\")]\n self._tool_cache = tool_cache\n self.tools = tool_list\n\n await logger.adebug(\n \"MCP update_tool_list: fetched from MCP count=%d server=%r\",\n len(tool_list),\n server_name,\n )\n\n # Cache the result only if caching is enabled\n if use_cache:\n cache_data = {\n \"tools\": tool_list,\n \"tool_names\": self.tool_names,\n \"tool_cache\": tool_cache,\n \"config\": server_config,\n }\n\n # Safely update the servers cache with bounded size (FIFO eviction).\n current_servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(current_servers_cache, dict):\n # Because the cache key now includes a header hash, a tenant that\n # rotates session tokens would grow this map without bound. Drop\n # the oldest entry when over the limit.\n max_entries = self.SHARED_SERVERS_CACHE_MAX_ENTRIES\n while (\n len(current_servers_cache) >= max_entries and servers_cache_key not in current_servers_cache\n ):\n oldest_key = next(iter(current_servers_cache))\n current_servers_cache.pop(oldest_key, None)\n current_servers_cache[servers_cache_key] = cache_data\n safe_cache_set(self._shared_component_cache, \"servers\", current_servers_cache)\n await logger.adebug(\n \"MCP update_tool_list: wrote shared_servers_cache key=%r size=%d\",\n servers_cache_key,\n len(current_servers_cache),\n )\n\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = (\n f\"Timeout updating tool list: {e!s}. \"\n \"Raise ``LANGFLOW_MCP_SERVER_TIMEOUT`` for the deployment if the MCP \"\n \"server legitimately needs more time to respond.\"\n )\n await logger.aexception(msg)\n raise TimeoutError(msg) from e\n except Exception as e:\n msg = f\"Error updating tool list: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return tool_list, {\"name\": server_name, \"config\": server_config}\n\n async def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None) -> dict:\n \"\"\"Toggle the visibility of connection-specific fields based on the selected mode.\"\"\"\n try:\n is_refresh = bool(build_config.get(\"is_refresh\", False))\n if field_name == \"tool\":\n try:\n # Always refresh tools when cache is disabled, or when tools list is empty\n # This ensures database edits are reflected immediately when cache is disabled\n use_cache = getattr(self, \"use_cache\", False)\n if is_refresh:\n self.tools = []\n if is_refresh or len(self.tools) == 0 or not use_cache:\n try:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n build_config[\"tool\"][\"options\"] = [tool.name for tool in self.tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout updating tool list: {e!s}\"\n await logger.aexception(msg)\n if not build_config[\"tools_metadata\"][\"show\"]:\n build_config[\"tool\"][\"show\"] = True\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = msg\n else:\n build_config[\"tool\"][\"show\"] = False\n except ValueError as e:\n msg = f\"Error updating tool list: {e!s}\"\n await logger.aexception(msg)\n if not build_config[\"tools_metadata\"][\"show\"]:\n build_config[\"tool\"][\"show\"] = True\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = msg\n else:\n build_config[\"tool\"][\"show\"] = False\n\n if field_value == \"\":\n return build_config\n tool_obj = None\n for tool in self.tools:\n if tool.name == field_value:\n tool_obj = tool\n break\n if tool_obj is None:\n msg = f\"Tool {field_value} not found in available tools: {self.tools}\"\n await logger.awarning(msg)\n return build_config\n await self._update_tool_config(build_config, field_value)\n except Exception as e:\n build_config[\"tool\"][\"options\"] = []\n msg = f\"Failed to update tools: {e!s}\"\n raise ValueError(msg) from e\n else:\n return build_config\n elif field_name == \"mcp_server\":\n if not field_value:\n build_config[\"tool\"][\"show\"] = False\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"value\"] = \"\"\n build_config[\"tool\"][\"placeholder\"] = \"\"\n build_config[\"tool_placeholder\"][\"tool_mode\"] = False\n self.remove_non_default_keys(build_config)\n return build_config\n\n build_config[\"tool_placeholder\"][\"tool_mode\"] = True\n\n current_server_name = field_value.get(\"name\") if isinstance(field_value, dict) else field_value\n build_config_server_value = build_config.get(\"mcp_server\", {}).get(\"value\")\n build_config_server_name = (\n build_config_server_value.get(\"name\")\n if isinstance(build_config_server_value, dict)\n else build_config_server_value\n )\n servers_cache_key_ui = self._mcp_servers_cache_key(current_server_name) if current_server_name else \"\"\n _last_selected_server = safe_cache_get(self._shared_component_cache, \"last_selected_server\", \"\")\n # Only treat as a server change if there was a previous server selection.\n # Cold cache (_last_selected_server=\"\") on initial flow load is NOT a server change —\n # the user didn't switch anything, the backend just hasn't seen this component yet.\n server_changed = bool(\n (_last_selected_server and current_server_name != _last_selected_server)\n or (build_config_server_name and current_server_name != build_config_server_name)\n )\n\n # Determine if \"Tool Mode\" is active by checking if the tool dropdown is hidden.\n is_in_tool_mode = build_config[\"tools_metadata\"][\"show\"]\n\n # Get use_cache setting to determine if we should use cached data\n use_cache = getattr(self, \"use_cache\", False)\n\n # Fast path: if server didn't change and we already have options, keep them as-is\n # BUT only if caching is enabled, we're in tool mode, or it's the initial load\n existing_options = build_config.get(\"tool\", {}).get(\"options\") or []\n if not is_refresh and not server_changed and existing_options:\n # In non-tool mode with cache disabled, skip the fast path to force refresh\n # BUT on initial load (cold cache), always preserve saved options from the flow\n if not is_in_tool_mode and not use_cache and _last_selected_server:\n pass # Continue to refresh logic below (user-initiated with cache disabled)\n else:\n if not is_in_tool_mode:\n build_config[\"tool\"][\"show\"] = True\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", current_server_name)\n return build_config\n\n # To avoid unnecessary updates, only proceed if the server has actually changed\n # OR if caching is disabled (to force refresh in non-tool mode)\n if (\n not is_refresh\n and (_last_selected_server in (current_server_name, \"\"))\n and build_config[\"tool\"][\"show\"]\n and use_cache\n ):\n if current_server_name:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict):\n cached = servers_cache.get(servers_cache_key_ui)\n if cached is not None and cached.get(\"tool_names\"):\n cached_tools = cached[\"tool_names\"]\n current_tools = build_config[\"tool\"][\"options\"]\n if current_tools == cached_tools:\n return build_config\n else:\n return build_config\n safe_cache_set(self._shared_component_cache, \"last_selected_server\", current_server_name)\n\n # When cache is disabled, clear any cached data for this server\n # This ensures we always fetch fresh data from the database\n if (is_refresh or not use_cache) and current_server_name:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict) and servers_cache_key_ui in servers_cache:\n servers_cache.pop(servers_cache_key_ui)\n safe_cache_set(self._shared_component_cache, \"servers\", servers_cache)\n\n # Check if tools are already cached for this server before clearing\n cached_tools = None\n if current_server_name and use_cache and not is_refresh:\n servers_cache = safe_cache_get(self._shared_component_cache, \"servers\", {})\n if isinstance(servers_cache, dict):\n cached = servers_cache.get(servers_cache_key_ui)\n if cached is not None:\n try:\n cached_tools = cached[\"tools\"]\n self.tools = cached_tools\n self.tool_names = cached[\"tool_names\"]\n self._tool_cache = cached[\"tool_cache\"]\n except (TypeError, KeyError, AttributeError) as e:\n # Handle corrupted cache data by ignoring it\n msg = f\"Unable to use cached data for MCP Server,{current_server_name}: {e}\"\n await logger.awarning(msg)\n cached_tools = None\n\n # Clear tools when cache is disabled OR when we don't have cached tools\n # This ensures fresh tools are fetched after database edits\n if is_refresh or not cached_tools or not use_cache:\n self.tools = [] # Clear previous tools to force refresh\n\n # Clear previous tool inputs if:\n # 1. Server actually changed\n # 2. Cache is disabled (meaning tool list will be refreshed)\n if is_refresh or server_changed or not use_cache:\n self.remove_non_default_keys(build_config)\n\n # Only show the tool dropdown if not in tool_mode\n if not is_in_tool_mode:\n build_config[\"tool\"][\"show\"] = True\n if cached_tools:\n # Use cached tools to populate options immediately\n build_config[\"tool\"][\"options\"] = [tool.name for tool in cached_tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n else:\n # Actually fetch tools now instead of deferring to a frontend callback.\n # The frontend has no reliable mechanism to trigger a second\n # update_build_config call for the \"tool\" field after this response,\n # so we must populate the options here.\n try:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list(\n mcp_server_value=field_value\n )\n build_config[\"tool\"][\"options\"] = [tool.name for tool in self.tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout loading tools for MCP server: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = \"Timeout on MCP server\"\n except (ValueError, ImportError, ConnectionError, OSError, RuntimeError) as e:\n msg = f\"Error loading tools for MCP server: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = (\n \"Error on MCP Server\" if \"'NoneType' object has no attribute 'id'\" in msg else msg\n )\n # Force a value refresh only when the user genuinely switched servers.\n # server_changed is only True for real user-initiated changes (not initial load).\n if server_changed:\n build_config[\"tool\"][\"value\"] = uuid.uuid4()\n else:\n # Keep the tool dropdown hidden if in tool_mode\n self._not_load_actions = True\n build_config[\"tool\"][\"show\"] = False\n\n elif field_name == \"tool_mode\":\n build_config[\"tool\"][\"placeholder\"] = \"\"\n build_config[\"tool\"][\"show\"] = not bool(field_value) and bool(build_config[\"mcp_server\"])\n self.remove_non_default_keys(build_config)\n self.tool = build_config[\"tool\"][\"value\"]\n if field_value:\n self._not_load_actions = True\n else:\n build_config[\"tool\"][\"value\"] = uuid.uuid4()\n build_config[\"tool\"][\"show\"] = True\n # Fetch tools immediately instead of showing \"Loading tools...\"\n try:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n build_config[\"tool\"][\"options\"] = [tool.name for tool in self.tools]\n build_config[\"tool\"][\"placeholder\"] = \"Select a tool\"\n except (TimeoutError, asyncio.TimeoutError) as e:\n msg = f\"Timeout loading tools when toggling tool mode: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = msg\n except (ValueError, ImportError, ConnectionError, OSError, RuntimeError) as e:\n msg = f\"Error loading tools when toggling tool mode: {e!s}\"\n await logger.aexception(msg)\n build_config[\"tool\"][\"options\"] = []\n build_config[\"tool\"][\"placeholder\"] = msg\n elif field_name == \"tool_execution_timeout\":\n try:\n val = float(field_value) if field_value not in (None, \"\") else 0.0\n except (ValueError, TypeError):\n val = 0.0\n if val < 0:\n build_config[\"tool_execution_timeout\"][\"placeholder\"] = (\n \"⚠ Value must be ≥ 0. Negative timeouts cause immediate failures.\"\n )\n build_config[\"tool_execution_timeout\"][\"value\"] = 0.0\n else:\n build_config[\"tool_execution_timeout\"][\"placeholder\"] = \"\"\n elif field_name == \"tools_metadata\":\n self._not_load_actions = False\n\n except Exception as e:\n msg = f\"Error in update_build_config: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n else:\n return build_config\n\n @staticmethod\n def _unwrap_optional_annotation(annotation: Any) -> Any:\n \"\"\"Remove a single None branch from a union annotation.\"\"\"\n if isinstance(annotation, UnionType):\n non_none = [item for item in get_args(annotation) if item is not type(None)]\n if len(non_none) == 1:\n return non_none[0]\n return annotation\n\n if get_origin(annotation) is None:\n return annotation\n\n non_none = [item for item in get_args(annotation) if item is not type(None)]\n if len(non_none) == 1 and len(non_none) != len(get_args(annotation)):\n return non_none[0]\n return annotation\n\n @classmethod\n def _is_object_like_annotation(cls, annotation: Any) -> bool:\n \"\"\"Return True when the annotation represents a dict-like payload.\"\"\"\n annotation = cls._unwrap_optional_annotation(annotation)\n origin = get_origin(annotation)\n if origin is dict:\n return True\n return annotation is dict or (isinstance(annotation, type) and issubclass(annotation, BaseModel))\n\n @classmethod\n def _should_include_tool_argument(cls, model_field: Any, value: Any) -> bool:\n \"\"\"Omit blank optional values so MCP server defaults remain intact.\"\"\"\n if value is None:\n return False\n\n if model_field.is_required():\n return True\n\n if isinstance(value, str) and value == \"\":\n return False\n\n return not (\n value == {} and model_field.default is None and cls._is_object_like_annotation(model_field.annotation)\n )\n\n def _build_tool_kwargs(self, args_schema: type[BaseModel]) -> dict[str, Any]:\n \"\"\"Collect tool kwargs from component inputs, omitting blank optional values.\"\"\"\n kwargs: dict[str, Any] = {}\n for arg_name, model_field in args_schema.model_fields.items():\n value = getattr(self, arg_name, None)\n if isinstance(value, Message):\n value = value.text\n\n if self._should_include_tool_argument(model_field, value):\n kwargs[arg_name] = value\n\n return kwargs\n\n def get_inputs_for_all_tools(self, tools: list) -> dict:\n \"\"\"Get input schemas for all tools.\"\"\"\n inputs = {}\n for tool in tools:\n if not tool or not hasattr(tool, \"name\"):\n continue\n try:\n langflow_inputs = schema_to_langflow_inputs(tool.args_schema)\n inputs[tool.name] = langflow_inputs\n except (AttributeError, ValueError, TypeError, KeyError) as e:\n msg = f\"Error getting inputs for tool {getattr(tool, 'name', 'unknown')}: {e!s}\"\n logger.exception(msg)\n continue\n return inputs\n\n def remove_non_default_keys(self, build_config: dict) -> None:\n \"\"\"Remove non-default keys from the build config.\"\"\"\n for key in list(build_config.keys()):\n if key not in self.default_keys:\n build_config.pop(key)\n\n async def _update_tool_config(self, build_config: dict, tool_name: str) -> None:\n \"\"\"Update tool configuration with proper error handling.\"\"\"\n if not self.tools:\n self.tools, build_config[\"mcp_server\"][\"value\"] = await self.update_tool_list()\n\n if not tool_name:\n return\n\n tool_obj = next((tool for tool in self.tools if tool.name == tool_name), None)\n if not tool_obj:\n msg = f\"Tool {tool_name} not found in available tools: {self.tools}\"\n self.remove_non_default_keys(build_config)\n build_config[\"tool\"][\"value\"] = \"\"\n await logger.awarning(msg)\n return\n\n try:\n # Store current values before removing inputs (only for the current tool)\n current_values = {}\n for key, value in build_config.items():\n if key not in self.default_keys and isinstance(value, dict) and \"value\" in value:\n current_values[key] = value[\"value\"]\n\n # Remove ALL non-default keys (all previous tool inputs)\n self.remove_non_default_keys(build_config)\n\n # Get and validate new inputs for the selected tool\n self.schema_inputs = await self._validate_schema_inputs(tool_obj)\n if not self.schema_inputs:\n msg = f\"No input parameters to configure for tool '{tool_name}'\"\n await logger.ainfo(msg)\n return\n\n # Add new inputs to build config for the selected tool only\n for schema_input in self.schema_inputs:\n if not schema_input or not hasattr(schema_input, \"name\"):\n msg = \"Invalid schema input detected, skipping\"\n await logger.awarning(msg)\n continue\n\n try:\n name = schema_input.name\n input_dict = schema_input.to_dict()\n input_dict.setdefault(\"value\", None)\n input_dict.setdefault(\"required\", True)\n\n build_config[name] = input_dict\n\n # Preserve existing value if the parameter name exists in current_values\n if name in current_values:\n build_config[name][\"value\"] = current_values[name]\n\n except (AttributeError, KeyError, TypeError) as e:\n msg = f\"Error processing schema input {schema_input}: {e!s}\"\n await logger.aexception(msg)\n continue\n except ValueError as e:\n msg = f\"Schema validation error for tool {tool_name}: {e!s}\"\n await logger.aexception(msg)\n self.schema_inputs = []\n return\n except (AttributeError, KeyError, TypeError) as e:\n msg = f\"Error updating tool config: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n\n async def build_output(self) -> DataFrame:\n \"\"\"Build output with improved error handling and validation.\"\"\"\n try:\n self.tools, _ = await self.update_tool_list()\n if self.tool != \"\":\n # Set session context for persistent MCP sessions using Langflow session ID\n session_context = self._get_session_context()\n if session_context:\n self.stdio_client.set_session_context(session_context)\n self.streamable_http_client.set_session_context(session_context)\n exec_tool = self._tool_cache[self.tool]\n kwargs = self._build_tool_kwargs(exec_tool.args_schema)\n unflattened_kwargs = maybe_unflatten_dict(kwargs)\n\n output = await exec_tool.coroutine(**unflattened_kwargs)\n tool_content = []\n for item in output.content:\n item_dict = item.model_dump()\n item_dict = self.process_output_item(item_dict)\n tool_content.append(item_dict)\n\n if isinstance(tool_content, list) and all(isinstance(x, dict) for x in tool_content):\n return DataFrame(tool_content)\n return DataFrame(data=tool_content)\n return DataFrame(data=[{\"error\": \"You must select a tool\"}])\n except Exception as e:\n msg = f\"Error in build_output: {e!s}\"\n await logger.aexception(msg)\n raise ValueError(msg) from e\n\n def process_output_item(self, item_dict):\n \"\"\"Process the output of a tool.\"\"\"\n if item_dict.get(\"type\") == \"text\":\n text = item_dict.get(\"text\")\n try:\n parsed = json.loads(text)\n # Ensure we always return a dictionary for DataFrame compatibility\n if isinstance(parsed, dict):\n return parsed\n # Wrap non-dict parsed values in a dictionary\n return {\"text\": text, \"parsed_value\": parsed, \"type\": \"text\"} # noqa: TRY300\n except json.JSONDecodeError:\n return item_dict\n return item_dict\n\n def _get_session_context(self) -> str | None:\n \"\"\"Get the Langflow session ID for MCP session caching.\"\"\"\n # Try to get session ID from the component's execution context\n if hasattr(self, \"graph\") and hasattr(self.graph, \"session_id\"):\n session_id = self.graph.session_id\n # Include server name to ensure different servers get different sessions\n server_name = \"\"\n mcp_server = getattr(self, \"mcp_server\", None)\n if isinstance(mcp_server, dict):\n server_name = mcp_server.get(\"name\", \"\")\n elif mcp_server:\n server_name = str(mcp_server)\n return f\"{session_id}_{server_name}\" if session_id else None\n return None\n\n async def _get_tools(self):\n \"\"\"Load tools for the agent toolkit; always refresh from ``update_tool_list``.\n\n ``_not_load_actions`` only applies to UI build_config flows (tool dropdown). Agent\n runs must always bind the current tool list (including after header/tweak changes).\n\n A short-lived TTL cache (``TOOL_TTL_SECS``, header-hash-keyed) skips the MCP\n round-trip when the same auth context is re-queried quickly (e.g. parallel agent\n steps sharing the same tweaked headers). The cache is per component instance and\n bounded by ``TOOL_TTL_MAX_ENTRIES``; it is distinct from the \"Use Cached Server\"\n (``use_cache``) toggle which controls the shared cross-request cache.\n \"\"\"\n mcp_server = getattr(self, \"mcp_server\", None)\n srv = mcp_server.get(\"name\") if isinstance(mcp_server, dict) else mcp_server\n\n ttl = self.TOOL_TTL_SECS\n ttl_key = self._mcp_servers_cache_key(srv or \"\") if ttl > 0 and srv else \"\"\n\n # TTL cache lookup + expired-entry eviction.\n if ttl > 0 and ttl_key:\n cached = self._ttl_tool_cache.get(ttl_key)\n if cached is not None:\n ts, cached_tools = cached\n age = time.monotonic() - ts\n if age < ttl:\n await logger.adebug(\n \"MCP _get_tools: TTL cache hit count=%d age=%.1fs server=%r\",\n len(cached_tools),\n age,\n srv,\n )\n return cached_tools\n # Stale — drop it so the size check below stays tight.\n self._ttl_tool_cache.pop(ttl_key, None)\n\n await logger.adebug(\"MCP _get_tools: fetching tool list server=%r\", srv)\n tools, _ = await self.update_tool_list(mcp_server)\n await logger.adebug(\"MCP _get_tools: fetched %d tools server=%r\", len(tools), srv)\n\n # TTL cache store with bounded size (FIFO eviction of the oldest entry).\n if ttl > 0 and ttl_key and tools:\n if len(self._ttl_tool_cache) >= self.TOOL_TTL_MAX_ENTRIES:\n # dict preserves insertion order — pop the oldest entry.\n oldest_key = next(iter(self._ttl_tool_cache))\n self._ttl_tool_cache.pop(oldest_key, None)\n self._ttl_tool_cache[ttl_key] = (time.monotonic(), tools)\n\n return tools\n" }, "headers": { "_input_type": "DictInput", @@ -2233,6 +2234,26 @@ "type": "str", "value": "remix_lock_layer" }, + "tool_execution_timeout": { + "_input_type": "FloatInput", + "advanced": true, + "display_name": "Tool Execution Timeout (seconds)", + "dynamic": false, + "info": "Maximum time to wait for tool execution before timing out. Supports decimal values for sub-second timeouts (e.g., 0.01 for 10ms). Set to 0 to use the system-configured MCP timeout.", + "list": false, + "list_add_label": "Add More", + "name": "tool_execution_timeout", + "override_skip": false, + "placeholder": "", + "required": false, + "show": true, + "title_case": false, + "tool_mode": false, + "trace_as_metadata": true, + "track_in_telemetry": true, + "type": "float", + "value": 0.0 + }, "tool_placeholder": { "_input_type": "MessageTextInput", "advanced": false, diff --git a/src/backend/base/langflow/locales/en.json b/src/backend/base/langflow/locales/en.json index 713d4aeba9..129f62df5a 100644 --- a/src/backend/base/langflow/locales/en.json +++ b/src/backend/base/langflow/locales/en.json @@ -4049,44 +4049,6 @@ "components.huggingfacemodel.inputs.typical_p.info.ed0e56c7": "Typical Decoding mass.", "components.huggingfacemodel.outputs.model_output.display_name.698eabeb": "Language Model", "components.huggingfacemodel.outputs.text_output.display_name.2544139a": "Model Response", - "components.ibmwatsonxmodel.description.4ae4048c": "Generate text using IBM watsonx.ai foundation models.", - "components.ibmwatsonxmodel.display_name.3cd17f91": "IBM watsonx.ai", - "components.ibmwatsonxmodel.inputs.api_key.display_name.c1368655": "Watsonx API Key", - "components.ibmwatsonxmodel.inputs.api_key.info.f48121a7": "The API Key to use for the model.", - "components.ibmwatsonxmodel.inputs.base_url.display_name.a882cd1d": "watsonx API Endpoint", - "components.ibmwatsonxmodel.inputs.base_url.info.ea1445bd": "The base URL of the API.", - "components.ibmwatsonxmodel.inputs.frequency_penalty.display_name.23155cbf": "Frequency Penalty", - "components.ibmwatsonxmodel.inputs.frequency_penalty.info.21188aa8": "Penalty for frequency of token usage.", - "components.ibmwatsonxmodel.inputs.input_value.display_name.36ecb4f8": "Input", - "components.ibmwatsonxmodel.inputs.logit_bias.display_name.0734e792": "Logit Bias", - "components.ibmwatsonxmodel.inputs.logit_bias.info.b7538c07": "JSON string of token IDs to bias or suppress (e.g., {\"1003\": -100, \"1004\": 100}).", - "components.ibmwatsonxmodel.inputs.logprobs.display_name.9488a8cd": "Log Probabilities", - "components.ibmwatsonxmodel.inputs.logprobs.info.cb94ab04": "Whether to return log probabilities of the output tokens.", - "components.ibmwatsonxmodel.inputs.max_tokens.display_name.c18e6c7f": "Max Tokens", - "components.ibmwatsonxmodel.inputs.max_tokens.info.dc26b19f": "The maximum number of tokens to generate.", - "components.ibmwatsonxmodel.inputs.model_name.display_name.a501a70a": "Model Name", - "components.ibmwatsonxmodel.inputs.presence_penalty.display_name.afe0af5f": "Presence Penalty", - "components.ibmwatsonxmodel.inputs.presence_penalty.info.01c276fd": "Penalty for token presence in prior text.", - "components.ibmwatsonxmodel.inputs.project_id.display_name.d1c47641": "watsonx Project_ID", - "components.ibmwatsonxmodel.inputs.project_id.info.1cf73803": "The project ID associated with the foundation model.", - "components.ibmwatsonxmodel.inputs.seed.display_name.c740e9da": "Random Seed", - "components.ibmwatsonxmodel.inputs.seed.info.9a5b508e": "The random seed for the model.", - "components.ibmwatsonxmodel.inputs.space_id.display_name.f5156dd8": "watsonx Space_ID", - "components.ibmwatsonxmodel.inputs.space_id.info.5695c3d0": "The deployment space ID associated with the foundation model.", - "components.ibmwatsonxmodel.inputs.stop_sequence.display_name.cd87248f": "Stop Sequence", - "components.ibmwatsonxmodel.inputs.stop_sequence.info.60195410": "Sequence where generation should stop.", - "components.ibmwatsonxmodel.inputs.stream.display_name.1eec3071": "Stream", - "components.ibmwatsonxmodel.inputs.stream.info.f835b3ff": "Stream the response from the model. Streaming works only in Chat.", - "components.ibmwatsonxmodel.inputs.system_message.display_name.8e5a3143": "System Message", - "components.ibmwatsonxmodel.inputs.system_message.info.149360fa": "System message to pass to the model.", - "components.ibmwatsonxmodel.inputs.temperature.display_name.b958ce8b": "Temperature", - "components.ibmwatsonxmodel.inputs.temperature.info.3497632c": "Controls randomness, higher values increase diversity.", - "components.ibmwatsonxmodel.inputs.top_logprobs.display_name.22a9dc31": "Top Log Probabilities", - "components.ibmwatsonxmodel.inputs.top_logprobs.info.d06c6c24": "Number of most likely tokens to return at each position.", - "components.ibmwatsonxmodel.inputs.top_p.display_name.c7a8872a": "Top P", - "components.ibmwatsonxmodel.inputs.top_p.info.45956258": "The cumulative probability cutoff for token selection. Lower values mean sampling from a smaller, more top-weighted nucleus.", - "components.ibmwatsonxmodel.outputs.model_output.display_name.698eabeb": "Language Model", - "components.ibmwatsonxmodel.outputs.text_output.display_name.2544139a": "Model Response", "components.idgenerator.description.050f38b7": "Generates a unique ID.", "components.idgenerator.display_name.717d5edc": "ID Generator", "components.idgenerator.inputs.unique_id.display_name.8e37953d": "Value", @@ -4567,6 +4529,8 @@ "components.mcptools.inputs.mcp_server.info.a40e8466": "Select the MCP Server that will be used by this component", "components.mcptools.inputs.tool.display_name.2e53bdcd": "Tool", "components.mcptools.inputs.tool.info.8999dc11": "Select the tool to execute", + "components.mcptools.inputs.tool_execution_timeout.display_name.48b62b89": "Tool Execution Timeout (seconds)", + "components.mcptools.inputs.tool_execution_timeout.info.3bff216b": "Maximum time to wait for tool execution before timing out. Supports decimal values for sub-second timeouts (e.g., 0.01 for 10ms). Set to 0 to use the system-configured MCP timeout.", "components.mcptools.inputs.tool_placeholder.display_name.6d01d1af": "Tool Placeholder", "components.mcptools.inputs.tool_placeholder.info.91cae561": "Placeholder for the tool", "components.mcptools.inputs.use_cache.display_name.950b319b": "Use Cached Server", @@ -6539,20 +6503,6 @@ "components.vlmruntranscription.inputs.timeout_seconds.display_name.1f966032": "Timeout (seconds)", "components.vlmruntranscription.inputs.timeout_seconds.info.1f8d294d": "Maximum time to wait for processing completion", "components.vlmruntranscription.outputs.result.display_name.6e7d50e8": "Result", - "components.watsonxembeddingscomponent.description.e5f3117c": "Generate embeddings using IBM watsonx.ai models.", - "components.watsonxembeddingscomponent.display_name.0b8a6a96": "IBM watsonx.ai Embeddings", - "components.watsonxembeddingscomponent.inputs.api_key.display_name.c1368655": "Watsonx API Key", - "components.watsonxembeddingscomponent.inputs.api_key.info.f48121a7": "The API Key to use for the model.", - "components.watsonxembeddingscomponent.inputs.input_text.display_name.8a7917fb": "Include the original text in the output", - "components.watsonxembeddingscomponent.inputs.model_name.display_name.a501a70a": "Model Name", - "components.watsonxembeddingscomponent.inputs.project_id.display_name.d1c47641": "watsonx Project_ID", - "components.watsonxembeddingscomponent.inputs.project_id.info.96ceeac2": "The project ID associated with the embedding model.", - "components.watsonxembeddingscomponent.inputs.space_id.display_name.f5156dd8": "watsonx Space_ID", - "components.watsonxembeddingscomponent.inputs.space_id.info.ccea7db1": "The deployment space ID associated with the embedding model.", - "components.watsonxembeddingscomponent.inputs.truncate_input_tokens.display_name.becdfebd": "Truncate Input Tokens", - "components.watsonxembeddingscomponent.inputs.url.display_name.a882cd1d": "watsonx API Endpoint", - "components.watsonxembeddingscomponent.inputs.url.info.ea1445bd": "The base URL of the API.", - "components.watsonxembeddingscomponent.outputs.embeddings.display_name.860ec100": "Embedding Model", "components.weaviate.description.fb9d3e1f": "Weaviate Vector Store with search capabilities", "components.weaviate.display_name.e95095e1": "Weaviate", "components.weaviate.inputs.api_key.display_name.23189d55": "API Key", diff --git a/src/backend/tests/unit/base/mcp/test_mcp_timeout_configuration.py b/src/backend/tests/unit/base/mcp/test_mcp_timeout_configuration.py new file mode 100644 index 0000000000..4e437459ef --- /dev/null +++ b/src/backend/tests/unit/base/mcp/test_mcp_timeout_configuration.py @@ -0,0 +1,402 @@ +"""Tests for MCP timeout configuration feature. + +This module tests the configurable timeout parameters added to support +long-running MCP tool executions (>30 seconds). + +Related to customer issue TS021996258 (Verizon). +""" + +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest +from lfx.base.mcp.util import MCPStdioClient, MCPStreamableHttpClient, update_tools + + +class TestMCPTimeoutConfiguration: + """Test timeout configuration at various levels.""" + + @pytest.mark.asyncio + async def test_stdio_client_default_timeout(self): + """Test that MCPStdioClient uses global default timeout (180s).""" + with patch("lfx.base.mcp.util._get_mcp_setting") as mock_get_setting: + # Simulate: mcp_tool_execution_timeout=180, mcp_server_timeout=20 + def get_setting_side_effect(key, default): + if key == "mcp_tool_execution_timeout": + return 180 + if key == "mcp_server_timeout": + return 20 + return default + + mock_get_setting.side_effect = get_setting_side_effect + + client = MCPStdioClient() + assert client._tool_execution_timeout == 180 + + @pytest.mark.asyncio + async def test_stdio_client_custom_timeout(self): + """Test that MCPStdioClient accepts custom timeout parameter.""" + client = MCPStdioClient(tool_execution_timeout=300) + assert client._tool_execution_timeout == 300 + + @pytest.mark.asyncio + async def test_streamable_http_client_default_timeout(self): + """Test that MCPStreamableHttpClient uses global default timeout (180s).""" + with patch("lfx.base.mcp.util._get_mcp_setting") as mock_get_setting: + # Simulate: mcp_tool_execution_timeout=180, mcp_server_timeout=20 + def get_setting_side_effect(key, default): + if key == "mcp_tool_execution_timeout": + return 180 + if key == "mcp_server_timeout": + return 20 + return default + + mock_get_setting.side_effect = get_setting_side_effect + + client = MCPStreamableHttpClient() + assert client._tool_execution_timeout == 180 + + @pytest.mark.asyncio + async def test_streamable_http_client_custom_timeout(self): + """Test that MCPStreamableHttpClient accepts custom timeout parameter.""" + client = MCPStreamableHttpClient(tool_execution_timeout=300) + assert client._tool_execution_timeout == 300 + + @pytest.mark.asyncio + async def test_stdio_run_tool_uses_client_timeout(self): + """Test that run_tool uses client's configured timeout.""" + client = MCPStdioClient(tool_execution_timeout=120) + client._connected = True + client._connection_params = {"command": "test"} + client._session_context = "test_context" + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value={"result": "success"}) + + with ( + patch.object(client, "_get_or_create_session", return_value=mock_session), + patch("asyncio.wait_for") as mock_wait_for, + ): + mock_wait_for.return_value = {"result": "success"} + await client.run_tool("test_tool", {"arg": "value"}) + + # Verify wait_for was called with client's timeout + assert mock_wait_for.call_count == 1 + call_args = mock_wait_for.call_args + assert call_args[1]["timeout"] == 120 + + @pytest.mark.asyncio + async def test_stdio_run_tool_timeout_override(self): + """Test that run_tool accepts per-call timeout override.""" + client = MCPStdioClient(tool_execution_timeout=120) + client._connected = True + client._connection_params = {"command": "test"} + client._session_context = "test_context" + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value={"result": "success"}) + + with ( + patch.object(client, "_get_or_create_session", return_value=mock_session), + patch("asyncio.wait_for") as mock_wait_for, + ): + mock_wait_for.return_value = {"result": "success"} + # Override with 300 seconds + await client.run_tool("test_tool", {"arg": "value"}, timeout=300) + + # Verify wait_for was called with override timeout + assert mock_wait_for.call_count == 1 + call_args = mock_wait_for.call_args + assert call_args[1]["timeout"] == 300 + + @pytest.mark.asyncio + async def test_streamable_http_run_tool_uses_client_timeout(self): + """Test that StreamableHttpClient run_tool uses client's configured timeout.""" + client = MCPStreamableHttpClient(tool_execution_timeout=150) + client._connected = True + client._connection_params = {"url": "http://test"} + client._session_context = "test_context" + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value={"result": "success"}) + + with ( + patch.object(client, "_get_or_create_session", return_value=mock_session), + patch("asyncio.wait_for") as mock_wait_for, + ): + mock_wait_for.return_value = {"result": "success"} + await client.run_tool("test_tool", {"arg": "value"}) + + # Verify wait_for was called with client's timeout + assert mock_wait_for.call_count == 1 + call_args = mock_wait_for.call_args + assert call_args[1]["timeout"] == 150 + + @pytest.mark.asyncio + async def test_streamable_http_run_tool_timeout_override(self): + """Test that StreamableHttpClient run_tool accepts per-call timeout override.""" + client = MCPStreamableHttpClient(tool_execution_timeout=150) + client._connected = True + client._connection_params = {"url": "http://test"} + client._session_context = "test_context" + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value={"result": "success"}) + + with ( + patch.object(client, "_get_or_create_session", return_value=mock_session), + patch("asyncio.wait_for") as mock_wait_for, + ): + mock_wait_for.return_value = {"result": "success"} + # Override with 400 seconds + await client.run_tool("test_tool", {"arg": "value"}, timeout=400) + + # Verify wait_for was called with override timeout + assert mock_wait_for.call_count == 1 + call_args = mock_wait_for.call_args + assert call_args[1]["timeout"] == 400 + + @pytest.mark.asyncio + async def test_update_tools_passes_timeout_to_stdio_client(self): + """Test that update_tools passes timeout when creating MCPStdioClient.""" + server_config = { + "mode": "Stdio", + "command": "test-command", + "args": [], + } + + with patch("lfx.base.mcp.util.MCPStdioClient") as mock_client_class: + mock_client = AsyncMock() + mock_client.connect_to_server = AsyncMock(return_value=[]) + mock_client._connected = True + mock_client_class.return_value = mock_client + + await update_tools( + server_name="test_server", + server_config=server_config, + tool_execution_timeout=250, + ) + + # Verify MCPStdioClient was created with timeout + mock_client_class.assert_called_once_with(tool_execution_timeout=250) + + @pytest.mark.asyncio + async def test_update_tools_passes_timeout_to_http_client(self): + """Test that update_tools passes timeout when creating MCPStreamableHttpClient.""" + server_config = { + "mode": "Streamable_HTTP", + "url": "http://test-server", + } + + with patch("lfx.base.mcp.util.MCPStreamableHttpClient") as mock_client_class: + mock_client = AsyncMock() + mock_client.connect_to_server = AsyncMock(return_value=[]) + mock_client._connected = True + mock_client_class.return_value = mock_client + + await update_tools( + server_name="test_server", + server_config=server_config, + tool_execution_timeout=350, + ) + + # Verify MCPStreamableHttpClient was created with timeout + mock_client_class.assert_called_once_with(tool_execution_timeout=350) + + +class TestMCPTimeoutBehavior: + """Test timeout behavior and error handling.""" + + @pytest.mark.asyncio + async def test_timeout_error_is_caught_and_retried(self): + """Test that timeout errors trigger retry logic.""" + client = MCPStdioClient(tool_execution_timeout=1) + client._connected = True + client._connection_params = {"command": "test"} + client._session_context = "test_context" + + mock_session = AsyncMock() + # First call times out, second succeeds + mock_session.call_tool = AsyncMock(side_effect=[asyncio.TimeoutError(), {"result": "success"}]) + + with ( + patch.object(client, "_get_or_create_session", return_value=mock_session), + patch("asyncio.wait_for") as mock_wait_for, + ): + # First call times out, second succeeds + mock_wait_for.side_effect = [asyncio.TimeoutError(), {"result": "success"}] + + result = await client.run_tool("test_tool", {"arg": "value"}) + + # Verify retry happened + assert mock_wait_for.call_count == 2 + assert result == {"result": "success"} + + @pytest.mark.asyncio + async def test_zero_timeout_uses_global_default(self): + """Test that timeout=0 or None falls back to global setting.""" + with patch("lfx.base.mcp.util._get_mcp_setting") as mock_get_setting: + # Simulate: mcp_tool_execution_timeout=180, mcp_server_timeout=20 + def get_setting_side_effect(key, default): + if key == "mcp_tool_execution_timeout": + return 180 + if key == "mcp_server_timeout": + return 20 + return default + + mock_get_setting.side_effect = get_setting_side_effect + + # Test with None - should fall back to global setting + client1 = MCPStdioClient(tool_execution_timeout=None) + assert client1._tool_execution_timeout == 180 + + # Test with 0 - should fall back to global setting since 0 is an invalid explicit timeout + client2 = MCPStdioClient(tool_execution_timeout=0) + assert client2._tool_execution_timeout == 180 + + @pytest.mark.asyncio + async def test_multiple_clients_independent_timeouts(self): + """Test that multiple client instances maintain independent timeout values.""" + # Create multiple clients with different timeout values + client1 = MCPStdioClient(tool_execution_timeout=100) + client2 = MCPStdioClient(tool_execution_timeout=200) + client3 = MCPStreamableHttpClient(tool_execution_timeout=300) + client4 = MCPStreamableHttpClient(tool_execution_timeout=400) + + # Verify each client maintains its own timeout value + assert client1._tool_execution_timeout == 100 + assert client2._tool_execution_timeout == 200 + assert client3._tool_execution_timeout == 300 + assert client4._tool_execution_timeout == 400 + + # Verify changing one doesn't affect others + client1._tool_execution_timeout = 150 + assert client1._tool_execution_timeout == 150 + assert client2._tool_execution_timeout == 200 # Unchanged + assert client3._tool_execution_timeout == 300 # Unchanged + assert client4._tool_execution_timeout == 400 # Unchanged + + @pytest.mark.asyncio + async def test_multiple_tool_calls_independent_timeout_overrides(self): + """Test that multiple tool calls with different timeout overrides remain independent.""" + client = MCPStdioClient(tool_execution_timeout=120) + client._connected = True + client._connection_params = {"command": "test"} + client._session_context = "test_context" + + mock_session = AsyncMock() + mock_session.call_tool = AsyncMock(return_value={"result": "success"}) + + with ( + patch.object(client, "_get_or_create_session", return_value=mock_session), + patch("asyncio.wait_for") as mock_wait_for, + ): + mock_wait_for.return_value = {"result": "success"} + + # Call multiple tools with different timeout overrides + await client.run_tool("tool1", {"arg": "value1"}, timeout=100) + await client.run_tool("tool2", {"arg": "value2"}, timeout=200) + await client.run_tool("tool3", {"arg": "value3"}, timeout=300) + await client.run_tool("tool4", {"arg": "value4"}) # Uses client default (120) + + # Verify each call used its own timeout + assert mock_wait_for.call_count == 4 + + # Check each call's timeout parameter + call_timeouts = [call[1]["timeout"] for call in mock_wait_for.call_args_list] + assert call_timeouts == [100, 200, 300, 120] + + @pytest.mark.asyncio + async def test_none_timeout_preserves_existing_client_timeout(self): + """Test that passing None as timeout preserves existing client's timeout.""" + from lfx.base.mcp.util import update_tools + + # Create a client with a specific timeout + initial_client = MCPStdioClient(tool_execution_timeout=250) + + server_config = { + "mode": "Stdio", + "command": "test-command", + "args": [], + } + + with patch.object(initial_client, "connect_to_server", new_callable=AsyncMock) as mock_connect: + mock_connect.return_value = [] + initial_client._connected = True + + # Call update_tools with None timeout - should preserve existing 250s + await update_tools( + server_name="test_server", + server_config=server_config, + tool_execution_timeout=None, + mcp_stdio_client=initial_client, + ) + + # Verify the client's timeout was NOT overwritten + assert initial_client._tool_execution_timeout == 250 + + @pytest.mark.asyncio + async def test_mcp_sse_client_receives_timeout(self): + """Test that mcp_sse_client (backward compatibility alias) receives timeout.""" + from lfx.base.mcp.util import update_tools + + # Create an SSE client (which is actually a StreamableHttpClient) + sse_client = MCPStreamableHttpClient(tool_execution_timeout=100) + + server_config = { + "mode": "Streamable_HTTP", + "url": "http://test-server", + } + + with patch.object(sse_client, "connect_to_server", new_callable=AsyncMock) as mock_connect: + mock_connect.return_value = [] + sse_client._connected = True + + # Call update_tools with mcp_sse_client and a new timeout + await update_tools( + server_name="test_server", + server_config=server_config, + tool_execution_timeout=350, + mcp_sse_client=sse_client, + ) + + # Verify the SSE client received the new timeout + assert sse_client._tool_execution_timeout == 350 + + +class TestMCPTimeoutSettingsValidation: + """Test MCP timeout validation that belongs with backend settings coverage.""" + + @pytest.mark.asyncio + async def test_global_setting_validation_rejects_zero(self): + """Test that global mcp_tool_execution_timeout setting rejects zero values.""" + from lfx.services.settings.base import Settings + + # Test the validator directly + with pytest.raises(ValueError, match="mcp_tool_execution_timeout must be greater than 0"): + Settings.validate_mcp_tool_execution_timeout(0.0) + + @pytest.mark.asyncio + async def test_global_setting_validation_rejects_negative(self): + """Test that global mcp_tool_execution_timeout setting rejects negative values.""" + from lfx.services.settings.base import Settings + + # Test the validator directly + with pytest.raises(ValueError, match="mcp_tool_execution_timeout must be greater than 0"): + Settings.validate_mcp_tool_execution_timeout(-100.0) + + @pytest.mark.asyncio + async def test_global_setting_accepts_positive_float(self): + """Test that global mcp_tool_execution_timeout setting accepts positive float values.""" + from lfx.services.settings.base import Settings + + # Test that the validator accepts positive values + # Note: We test the validator logic, not the full Settings initialization + validated_value = Settings.validate_mcp_tool_execution_timeout(180.0) + assert validated_value == 180.0 + + validated_value = Settings.validate_mcp_tool_execution_timeout(250.5) + assert validated_value == 250.5 + + validated_value = Settings.validate_mcp_tool_execution_timeout(0.5) + assert validated_value == 0.5 diff --git a/src/lfx/src/lfx/base/mcp/util.py b/src/lfx/src/lfx/base/mcp/util.py index d9d0f90da8..bdfe260c28 100644 --- a/src/lfx/src/lfx/base/mcp/util.py +++ b/src/lfx/src/lfx/base/mcp/util.py @@ -55,6 +55,28 @@ def _get_mcp_setting(key: str, default: Any = None) -> Any: return _mcp_settings_cache[key] +def _resolve_mcp_tool_execution_timeout(tool_execution_timeout: float | None) -> float: + """Resolve MCP tool execution timeout from explicit input or MCP settings. + + Priority for picking the timeout: + 1. `tool_execution_timeout`: Custom UI override directly on the component (Highest). + 2. Global Settings: The maximum value between `mcp_tool_execution_timeout` + and `mcp_server_timeout` from global settings. + 3. Fallback: 180.0 seconds if no custom or global settings exist (Lowest). + + Negative values are treated as unset (use system default) because + ``asyncio.wait_for`` immediately raises ``TimeoutError`` for any timeout < 0. + """ + if tool_execution_timeout is not None and float(tool_execution_timeout) > 0: + return float(tool_execution_timeout) + + configured = _get_mcp_setting("mcp_tool_execution_timeout", None) + mcp_server_timeout = _get_mcp_setting("mcp_server_timeout", None) + + configured_timeouts = [float(value) for value in (configured, mcp_server_timeout) if value is not None] + return max(configured_timeouts) if configured_timeouts else 180.0 + + def get_max_sessions_per_server() -> int: """Get maximum number of sessions per server to prevent resource exhaustion.""" return _get_mcp_setting("mcp_max_sessions_per_server") @@ -1548,12 +1570,13 @@ class MCPSessionManager: class MCPStdioClient: - def __init__(self, component_cache=None): + def __init__(self, component_cache=None, tool_execution_timeout: float | None = None): self.session: ClientSession | None = None self._connection_params = None self._connected = False self._session_context: str | None = None self._component_cache = component_cache + self._tool_execution_timeout = _resolve_mcp_tool_execution_timeout(tool_execution_timeout) async def _connect_to_server(self, command_str: str, env: dict[str, str] | None = None) -> list[StructuredTool]: """Connect to MCP server using stdio transport (SDK style). @@ -1647,12 +1670,13 @@ class MCPStdioClient: session_manager = self._get_session_manager() return await session_manager.get_session(self._session_context, self._connection_params, "stdio") - async def run_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: + async def run_tool(self, tool_name: str, arguments: dict[str, Any], timeout: float | None = None) -> Any: # noqa: ASYNC109 """Run a tool with the given arguments using context-specific session. Args: tool_name: Name of the tool to run arguments: Dictionary of arguments to pass to the tool + timeout: Optional timeout in seconds. If not provided, uses the client's configured timeout. Returns: The result of the tool execution @@ -1672,9 +1696,9 @@ class MCPStdioClient: param_hash = uuid.uuid4().hex[:8] self._session_context = f"default_{param_hash}" - # Tool-call timeout: env LANGFLOW_MCP_SERVER_TIMEOUT (via settings), with a 180s - # floor so default deployments aren't shorter than the previous hardcoded 30s. - timeout = max(get_settings_service().settings.mcp_server_timeout, 180.0) + # Use provided timeout or fall back to client's configured timeout + effective_timeout = timeout if timeout is not None else self._tool_execution_timeout + max_retries = 2 last_error_type = None @@ -1686,7 +1710,7 @@ class MCPStdioClient: result = await asyncio.wait_for( session.call_tool(tool_name, arguments=arguments), - timeout=timeout, + timeout=effective_timeout, ) except Exception as e: current_error_type = type(e).__name__ @@ -1780,12 +1804,13 @@ class MCPStdioClient: class MCPStreamableHttpClient: - def __init__(self, component_cache=None): + def __init__(self, component_cache=None, tool_execution_timeout: float | None = None): self.session: ClientSession | None = None self._connection_params = None self._connected = False self._session_context: str | None = None self._component_cache = component_cache + self._tool_execution_timeout = _resolve_mcp_tool_execution_timeout(tool_execution_timeout) def _get_session_manager(self) -> MCPSessionManager: """Get or create session manager from component cache.""" @@ -1930,12 +1955,13 @@ class MCPStreamableHttpClient: # DELETE is advisory—log and continue logger.debug(f"Unable to send session DELETE to '{url}': {e}") - async def run_tool(self, tool_name: str, arguments: dict[str, Any]) -> Any: + async def run_tool(self, tool_name: str, arguments: dict[str, Any], timeout: float | None = None) -> Any: # noqa: ASYNC109 """Run a tool with the given arguments using context-specific session. Args: tool_name: Name of the tool to run arguments: Dictionary of arguments to pass to the tool + timeout: Optional timeout in seconds. If not provided, uses the client's configured timeout. Returns: The result of the tool execution @@ -1955,9 +1981,9 @@ class MCPStreamableHttpClient: param_hash = uuid.uuid4().hex[:8] self._session_context = f"default_http_{param_hash}" - # Tool-call timeout: env LANGFLOW_MCP_SERVER_TIMEOUT (via settings), with a 180s - # floor so default deployments aren't shorter than the previous hardcoded 30s. - timeout = max(get_settings_service().settings.mcp_server_timeout, 180.0) + # Use provided timeout or fall back to client's configured timeout + effective_timeout = timeout if timeout is not None else self._tool_execution_timeout + max_retries = 2 last_error_type = None @@ -1969,7 +1995,7 @@ class MCPStreamableHttpClient: result = await asyncio.wait_for( session.call_tool(tool_name, arguments=arguments), - timeout=timeout, + timeout=effective_timeout, ) except Exception as e: current_error_type = type(e).__name__ @@ -2065,6 +2091,7 @@ async def update_tools( mcp_streamable_http_client: MCPStreamableHttpClient | None = None, mcp_sse_client: MCPStreamableHttpClient | None = None, # Backward compatibility request_variables: dict[str, str] | None = None, + tool_execution_timeout: float | None = None, ) -> tuple[str, list[StructuredTool], dict[str, StructuredTool]]: """Fetch server config and update available tools. @@ -2075,17 +2102,35 @@ async def update_tools( mcp_streamable_http_client: Optional streamable HTTP client instance mcp_sse_client: Optional SSE client instance (backward compatibility) request_variables: Optional dict of global variables to resolve in headers + tool_execution_timeout: Optional timeout in seconds for tool execution (int or float) """ if server_config is None: server_config = {} if not server_name: return "", [], {} + if mcp_stdio_client is None: - mcp_stdio_client = MCPStdioClient() + mcp_stdio_client = MCPStdioClient(tool_execution_timeout=tool_execution_timeout) + # Update timeout on existing client only if a new timeout is provided. + # Route through _resolve_mcp_tool_execution_timeout so that negative values + # (entered before UI validation fires) never reach asyncio.wait_for. + elif tool_execution_timeout is not None: + mcp_stdio_client._tool_execution_timeout = _resolve_mcp_tool_execution_timeout(tool_execution_timeout) # Backward compatibility: accept mcp_sse_client parameter if mcp_streamable_http_client is None: - mcp_streamable_http_client = mcp_sse_client if mcp_sse_client is not None else MCPStreamableHttpClient() + if mcp_sse_client is not None: + mcp_streamable_http_client = mcp_sse_client + # Set timeout on the aliased client if provided + if tool_execution_timeout is not None: + mcp_streamable_http_client._tool_execution_timeout = _resolve_mcp_tool_execution_timeout( + tool_execution_timeout + ) + else: + mcp_streamable_http_client = MCPStreamableHttpClient(tool_execution_timeout=tool_execution_timeout) + # Update timeout on existing client only if a new timeout is provided + elif tool_execution_timeout is not None: + mcp_streamable_http_client._tool_execution_timeout = _resolve_mcp_tool_execution_timeout(tool_execution_timeout) # Fetch server config from backend # Determine mode from config, defaulting to Streamable_HTTP if URL present diff --git a/src/lfx/src/lfx/components/models_and_agents/mcp_component.py b/src/lfx/src/lfx/components/models_and_agents/mcp_component.py index 09fed4ca2d..4a50aa90dc 100644 --- a/src/lfx/src/lfx/components/models_and_agents/mcp_component.py +++ b/src/lfx/src/lfx/components/models_and_agents/mcp_component.py @@ -20,7 +20,7 @@ from lfx.base.mcp.util import ( from lfx.base.tools.constants import TOOL_OUTPUT_DISPLAY_NAME, TOOL_OUTPUT_NAME from lfx.custom.custom_component.component_with_cache import ComponentWithCache from lfx.inputs.inputs import InputTypes # noqa: TC001 -from lfx.io import BoolInput, DictInput, DropdownInput, McpInput, MessageTextInput, Output +from lfx.io import BoolInput, DictInput, DropdownInput, FloatInput, McpInput, MessageTextInput, Output from lfx.io.schema import schema_to_langflow_inputs from lfx.log.logger import logger from lfx.schema.dataframe import DataFrame @@ -99,7 +99,8 @@ class MCPToolsComponent(ComponentWithCache): # Initialize cache keys to avoid CacheMiss when accessing them self._ensure_cache_structure() - # Initialize clients with access to the component cache + # Initialize clients with access to the component cache. + # Per-component timeout is normalized and applied immediately before MCP calls. self.stdio_client: MCPStdioClient = MCPStdioClient(component_cache=self._shared_component_cache) self.streamable_http_client: MCPStreamableHttpClient = MCPStreamableHttpClient( component_cache=self._shared_component_cache @@ -138,14 +139,48 @@ class MCPToolsComponent(ComponentWithCache): return {str(k): str(v) for k, v in component_headers.items()} return {} + def _normalize_tool_execution_timeout(self) -> float | None: + """Normalize the timeout input and reject negative values with a field-specific error.""" + timeout_value = getattr(self, "tool_execution_timeout", 0.0) + + if timeout_value in (None, ""): + return None + + try: + val = float(timeout_value) + except (ValueError, TypeError): + return None + + if val < 0: + msg = "Tool Execution Timeout must be greater than or equal to 0." + raise ValueError(msg) + + return val if val else None + def _mcp_servers_cache_key(self, server_name: str) -> str: - """Cache key for shared servers map; includes headers so auth/tweak changes get distinct entries.""" + """Cache key for shared servers map. + + Includes headers and timeout so auth/tweak/timeout changes get distinct entries. + """ if not server_name: return "" + + raw_timeout = getattr(self, "tool_execution_timeout", 0.0) or 0.0 + normalized_timeout = max(0.0, float(raw_timeout)) + hdrs = self._normalized_headers_for_cache() - if not hdrs: + + # Build cache key components + cache_data = { + "headers": hdrs, + "timeout": normalized_timeout, + } + + # If no headers and default timeout, just use server name + if not hdrs and normalized_timeout == 0.0: return server_name - payload = json.dumps(hdrs, sort_keys=True) + + payload = json.dumps(cache_data, sort_keys=True) digest = hashlib.sha256(payload.encode()).hexdigest()[:16] return f"{server_name}:{digest}" @@ -184,6 +219,7 @@ class MCPToolsComponent(ComponentWithCache): "use_cache", "verify_ssl", "headers", + "tool_execution_timeout", ] display_name = "MCP Tools" @@ -230,6 +266,19 @@ class MCPToolsComponent(ComponentWithCache): advanced=True, is_list=True, ), + FloatInput( + name="tool_execution_timeout", + display_name="Tool Execution Timeout (seconds)", + info=( + "Maximum time to wait for tool execution before timing out. " + "Supports decimal values for sub-second timeouts (e.g., 0.01 for 10ms). " + "Set to 0 to use the system-configured MCP timeout." + ), + value=0.0, + range_spec={"min": 0.0, "max": 3600.0, "step": 0.01}, + real_time_refresh=True, + advanced=True, + ), DropdownInput( name="tool", display_name="Tool", @@ -460,12 +509,15 @@ class MCPToolsComponent(ComponentWithCache): else "list-or-empty", ) + timeout = self._normalize_tool_execution_timeout() + _, tool_list, tool_cache = await update_tools( server_name=server_name, server_config=server_config, mcp_stdio_client=self.stdio_client, mcp_streamable_http_client=self.streamable_http_client, request_variables=request_variables, + tool_execution_timeout=timeout, ) self.tool_names = [tool.name for tool in tool_list if hasattr(tool, "name")] @@ -748,6 +800,18 @@ class MCPToolsComponent(ComponentWithCache): await logger.aexception(msg) build_config["tool"]["options"] = [] build_config["tool"]["placeholder"] = msg + elif field_name == "tool_execution_timeout": + try: + val = float(field_value) if field_value not in (None, "") else 0.0 + except (ValueError, TypeError): + val = 0.0 + if val < 0: + build_config["tool_execution_timeout"]["placeholder"] = ( + "⚠ Value must be ≥ 0. Negative timeouts cause immediate failures." + ) + build_config["tool_execution_timeout"]["value"] = 0.0 + else: + build_config["tool_execution_timeout"]["placeholder"] = "" elif field_name == "tools_metadata": self._not_load_actions = False diff --git a/src/lfx/src/lfx/services/settings/base.py b/src/lfx/src/lfx/services/settings/base.py index 2256fc9685..a55fc404fd 100644 --- a/src/lfx/src/lfx/services/settings/base.py +++ b/src/lfx/src/lfx/services/settings/base.py @@ -126,8 +126,23 @@ class Settings(BaseSettings): the browser's window.location.origin.""" mcp_server_timeout: int = 20 - """The number of seconds to wait before giving up on a lock to released or establishing a connection to the - database.""" + """The number of seconds to wait before giving up on establishing a connection to the MCP server.""" + + mcp_tool_execution_timeout: float = 180.0 + """Maximum seconds to wait for MCP tool execution before timing out. + Default is 180 seconds (3 minutes) to support long-running operations. + Supports decimal values for sub-second timeouts (e.g., 0.5 for 500ms). + Individual components can override this with their own timeout setting. + Must be a positive number greater than 0.""" + + @field_validator("mcp_tool_execution_timeout") + @classmethod + def validate_mcp_tool_execution_timeout(cls, v: float) -> float: + """Validate that mcp_tool_execution_timeout is positive.""" + if v <= 0: + msg = "mcp_tool_execution_timeout must be greater than 0" + raise ValueError(msg) + return v # --------------------------------------------------------------------- # MCP Session-manager tuning diff --git a/src/lfx/tests/unit/inputs/test_inputs_schema.py b/src/lfx/tests/unit/inputs/test_inputs_schema.py index 8254ab93b4..29a32f6d43 100644 --- a/src/lfx/tests/unit/inputs/test_inputs_schema.py +++ b/src/lfx/tests/unit/inputs/test_inputs_schema.py @@ -322,6 +322,21 @@ def test_schema_to_langflow_inputs_preserves_optional_defaults_and_nullable_obje assert inputs["proxy_country"].value == "us" +def test_float_input_allows_range_spec_minimum_for_non_negative_values(): + input_field = FloatInput.model_validate( + { + "name": "tool_execution_timeout", + "value": 0.0, + "range_spec": {"min": 0.0, "max": 3600.0, "step": 0.01}, + } + ) + + assert input_field.range_spec is not None + assert input_field.range_spec.min == 0.0 + assert input_field.range_spec.max == 3600.0 + assert input_field.range_spec.step == 0.01 + + def test_schema_to_langflow_inputs_invalid_type(): # Define a schema with an unsupported type class CustomType: diff --git a/src/lfx/tests/unit/mcp/test_mcp_component_timeout_validation.py b/src/lfx/tests/unit/mcp/test_mcp_component_timeout_validation.py new file mode 100644 index 0000000000..c4354f128b --- /dev/null +++ b/src/lfx/tests/unit/mcp/test_mcp_component_timeout_validation.py @@ -0,0 +1,58 @@ +import pytest +from lfx.components.models_and_agents.mcp_component import MCPToolsComponent + + +def test_normalize_tool_execution_timeout_returns_none_for_zero(): + component = MCPToolsComponent() + + component.tool_execution_timeout = 0.0 + + assert component._normalize_tool_execution_timeout() is None + + +def test_normalize_tool_execution_timeout_returns_float_for_positive_value(): + component = MCPToolsComponent() + + component.tool_execution_timeout = 12 + + assert component._normalize_tool_execution_timeout() == 12.0 + + +def test_normalize_tool_execution_timeout_rejects_negative_value_with_field_specific_message(): + component = MCPToolsComponent() + + component.tool_execution_timeout = -1 + + with pytest.raises(ValueError, match=r"Tool Execution Timeout must be greater than or equal to 0\."): + component._normalize_tool_execution_timeout() + + +def test_mcp_servers_cache_key_clamps_negative_timeout_to_zero(): + component = MCPToolsComponent() + component.tool_execution_timeout = -1 + + cache_key = component._mcp_servers_cache_key("demo-server") + assert "demo-server" in cache_key + + +@pytest.mark.asyncio +async def test_update_build_config_resets_negative_timeout(): + component = MCPToolsComponent() + build_config = {"tool_execution_timeout": {"value": -1, "placeholder": ""}} + + result = await component.update_build_config(build_config, "-1", "tool_execution_timeout") + + assert result["tool_execution_timeout"]["value"] == 0.0 + assert "Negative timeouts cause immediate failures" in result["tool_execution_timeout"]["placeholder"] + + +@pytest.mark.asyncio +async def test_update_build_config_handles_empty_timeout_safely(): + component = MCPToolsComponent() + build_config = {"tool_execution_timeout": {"value": "", "placeholder": ""}} + + # Should not crash and should clear any placeholder + result = await component.update_build_config(build_config, "", "tool_execution_timeout") + + assert result["tool_execution_timeout"]["placeholder"] == "" + assert result["tool_execution_timeout"]["value"] == "" diff --git a/src/lfx/tests/unit/mcp/test_tool_execution_timeout.py b/src/lfx/tests/unit/mcp/test_tool_execution_timeout.py new file mode 100644 index 0000000000..2c025f126e --- /dev/null +++ b/src/lfx/tests/unit/mcp/test_tool_execution_timeout.py @@ -0,0 +1,72 @@ +from unittest.mock import patch + +from lfx.base.mcp.util import ( + MCPStdioClient, + MCPStreamableHttpClient, + _resolve_mcp_tool_execution_timeout, +) + + +def test_resolve_mcp_tool_execution_timeout_uses_explicit_value(): + assert _resolve_mcp_tool_execution_timeout(42) == 42.0 + + +def test_resolve_mcp_tool_execution_timeout_uses_max_of_global_settings(): + with patch("lfx.base.mcp.util._get_mcp_setting") as mock_get_mcp_setting: + mock_get_mcp_setting.side_effect = lambda key, default=None: { + "mcp_tool_execution_timeout": 120.0, + "mcp_server_timeout": 240, + }.get(key, default) + + assert _resolve_mcp_tool_execution_timeout(None) == 240.0 + + +def test_resolve_mcp_tool_execution_timeout_uses_tool_timeout_when_larger(): + with patch("lfx.base.mcp.util._get_mcp_setting") as mock_get_mcp_setting: + mock_get_mcp_setting.side_effect = lambda key, default=None: { + "mcp_tool_execution_timeout": 300.0, + "mcp_server_timeout": 20, + }.get(key, default) + + assert _resolve_mcp_tool_execution_timeout(None) == 300.0 + + +def test_resolve_mcp_tool_execution_timeout_uses_server_timeout_when_tool_timeout_is_missing(): + with patch("lfx.base.mcp.util._get_mcp_setting") as mock_get_mcp_setting: + mock_get_mcp_setting.side_effect = lambda key, default=None: { + "mcp_tool_execution_timeout": None, + "mcp_server_timeout": 240, + }.get(key, default) + + assert _resolve_mcp_tool_execution_timeout(None) == 240.0 + + +def test_resolve_mcp_tool_execution_timeout_uses_server_timeout_when_tool_timeout_is_not_set(): + with patch("lfx.base.mcp.util._get_mcp_setting") as mock_get_mcp_setting: + mock_get_mcp_setting.side_effect = lambda key, default=None: { + "mcp_server_timeout": 210, + }.get(key, default) + + assert _resolve_mcp_tool_execution_timeout(None) == 210.0 + + +def test_resolve_mcp_tool_execution_timeout_falls_back_to_180_when_no_settings_exist(): + with patch("lfx.base.mcp.util._get_mcp_setting", return_value=None): + assert _resolve_mcp_tool_execution_timeout(None) == 180.0 + + +def test_mcp_stdio_client_uses_resolved_timeout(): + with patch("lfx.base.mcp.util._resolve_mcp_tool_execution_timeout", return_value=240.0): + client = MCPStdioClient(tool_execution_timeout=None) + + assert client._tool_execution_timeout == 240.0 + + +def test_mcp_streamable_http_client_uses_resolved_timeout(): + with patch("lfx.base.mcp.util._resolve_mcp_tool_execution_timeout", return_value=240.0): + client = MCPStreamableHttpClient(tool_execution_timeout=None) + + assert client._tool_execution_timeout == 240.0 + + +# Made with Bob