From cf9151729c571e2d63cee8ad4d290095eb4e588b Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Thu, 25 Jun 2026 15:42:42 +0000 Subject: [PATCH] [autofix.ci] apply automated fixes (attempt 2/3) --- .../langflow/initial_setup/starter_projects/Nvidia Remix.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 1e0883dbed..dc9285f766 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 @@ -2108,7 +2108,7 @@ "legacy": false, "lf_version": "1.4.2", "metadata": { - "code_hash": "c7652a63f23e", + "code_hash": "ba6770d8a484", "dependencies": { "dependencies": [ { @@ -2174,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, 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" + "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 # ``get_all_decrypted_variables`` is DB-service-only; the minimal\n # lfx VariableService (now registered by ``lfx serve``) lacks it, so\n # skip cleanly rather than raising into the broad except below.\n if variable_service and hasattr(variable_service, \"get_all_decrypted_variables\"):\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",