fix: mcp component dyanamic tool (#12779)

* fix(mcp): support real-time tool onboarding via per-request auth headers

Problem
-------
When Langflow runs an agent that has an MCPTools component attached and an
API request supplies authentication headers via tweaks, the MCP server can
legitimately return an *expanded* tool list that is broader than the
unauthenticated base set (the server filters its listing by the caller's
identity). Today those extra tools never reach the agent at runtime: the
component always binds the first preloaded tool list and ignores the new
auth context for the lifetime of the flow.

Root cause
----------
Several caching / binding layers combine to prevent per-request tool
refresh:

1. Flow JSON persists ``component_as_tool`` with ``cache: true`` and
   Langflow's ``Output`` memoization reuses the first ``to_toolkit()``
   result for all subsequent requests regardless of headers.
2. The shared MCP server cache was keyed by server name only, so
   requests with different auth contexts collided on the same cache slot.
3. In tool-mode, ``_get_tools()`` returned ``[]`` when
   ``_not_load_actions`` was true, starving the agent of MCP tools.
4. Concurrent ``update_tool_list`` calls raced on the same Streamable
   HTTP client session, producing intermittent HTTP 404 /
   "Session terminated" errors from the MCP SDK.

What changed
------------
``src/lfx/src/lfx/components/models_and_agents/mcp_component.py``:

- FIX 1 - Output cache bypass: ``_build_tool_output`` declares the
  Toolset output with ``cache=False``, and ``map_outputs`` overrides any
  ``cache: true`` value persisted in saved flow JSON. Together these
  guarantee every run resolves a fresh ``to_toolkit()`` call regardless
  of whether the flow was loaded from disk. Independent of the
  user-facing ``use_cache`` / "Use Cached Server" toggle.

- FIX 2 - Header-aware cache key: ``_mcp_servers_cache_key`` now hashes
  the component headers into the shared server-cache key so requests
  with different auth contexts land in distinct cache slots instead of
  masking each other. The shared cache is bounded at
  ``SHARED_SERVERS_CACHE_MAX_ENTRIES`` = 64 with FIFO eviction so a
  tenant rotating session tokens does not grow the map without limit.

- FIX 3 - ``_get_tools`` always fetches: removed the ``_not_load_actions``
  short-circuit that returned ``[]`` in tool-mode, so the agent always
  binds the current (possibly expanded) tool list. ``_not_load_actions``
  still gates only the UI build_config dropdown.

- FIX 4 - Serialized ``update_tool_list``: an ``asyncio.Lock`` prevents
  concurrent refreshes from racing on the same Streamable HTTP client
  session, eliminating the intermittent HTTP 404 / "Session terminated"
  errors the MCP SDK raised when session DELETE and POST overlapped.

- FIX 5 - TTL tool cache: ``_get_tools`` keeps a per-instance,
  header-hash-keyed TTL cache (``TOOL_TTL_SECS`` = 30s, disable with 0)
  bounded at ``TOOL_TTL_MAX_ENTRIES`` = 32 with FIFO eviction and
  stale-on-read drop. Parallel agent steps that share the same auth
  reuse the tool list instead of each paying for a fresh MCP round-trip.
  This is deliberately distinct from ``use_cache`` / "Use Cached Server"
  which controls the cross-request shared cache. The dict is initialized
  in ``__init__`` so every instance gets its own cache.

Impact on users
---------------
- Backward compatible. Unauthenticated / non-tweaked flows behave
  identically: the TTL cache is per-instance and scoped to a single
  (server, header-hash) pair, and the base-class ``to_toolkit`` chain is
  unchanged (no override of filtering via ``tools_metadata`` /
  ``enabled_tools``).
- New behaviour only triggers when tweaks/UI headers include an auth
  context; the header-hash cache key then isolates per-caller tool lists
  and the agent binds the correctly-filtered expanded set.
- No changes to ``lfx/services/settings/base.py``: the global
  ``mcp_server_timeout`` default stays at its existing value.

Test plan
---------
- ``python -m py_compile`` passes on the edited module.
- ``python -m ruff check`` reports no new violations introduced by this
  change.
- Local verification against an MCP server that returns different tool
  lists per caller identity:
  * Unauthenticated request returns the base tools as before.
  * Authenticated request (auth headers via tweaks) returns the
    expanded tool set on the first call and on subsequent calls with
    the same auth - TTL cache hits log
    ``MCP _get_tools: TTL cache hit``.
  * Switching to a different auth context on the same flow returns the
    correct per-caller tool list (header-hash cache key isolates the
    two contexts).
  * Parallel agent runs no longer surface HTTP 404 / "Session
    terminated" from the MCP SDK.
  * Disabling a tool via the UI (``tools_metadata``) correctly hides it
    from the bound agent - base-class filtering is preserved.

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* [autofix.ci] apply automated fixes (attempt 3/3)

* test(mcp): cover dynamic-tool onboarding fixes

Add unit coverage for the fixes introduced in this PR so regressions in the
cache-key, TTL cache, concurrency lock, shared-cache eviction, and Toolset
Output.cache=False behaviour are caught by CI instead of by users:

- Header normalization across list / dict / None / malformed shapes
- `_mcp_servers_cache_key` determinism + header-hash scoping across auth
  contexts (different headers → different keys; order-independent)
- Per-instance `_ttl_tool_cache` isolation (no cross-instance leakage)
- `_get_tools` TTL cache: hit skips `update_tool_list`, expired entries are
  refetched, FIFO-eviction caps growth at `TOOL_TTL_MAX_ENTRIES`, and TTL=0
  disables the cache
- `_update_tool_list_lock` serialises concurrent calls (peak concurrency 1)
- Shared `servers` cross-request cache evicts oldest entry when the map
  reaches `SHARED_SERVERS_CACHE_MAX_ENTRIES`
- `_build_tool_output` declares `cache=False`; `map_outputs` overrides
  persisted `cache=True` from saved flow JSON

* [autofix.ci] apply automated fixes

---------

Co-authored-by: Mansura <mansura.nw@gmail.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Eric Hare <ericrhare@gmail.com>

src/backend/tests/unit/components/models_and_agents/test_mcp_component_dynamic.py

@@ -0,0 +1,319 @@
+"""Unit tests for the dynamic tool-onboarding fixes in MCPToolsComponent.
+
+Covers:
+- ``_normalized_headers_for_cache`` handles list / dict / None header shapes
+- ``_mcp_servers_cache_key`` is deterministic, header-hash scoped, and distinguishes auth contexts
+- ``_ttl_tool_cache`` is a per-instance dict (not a class-level shared dict) so distinct
+  components cannot leak tool lists into each other
+- ``_get_tools`` honours the TTL cache hit / expiry / FIFO-eviction logic
+- ``update_tool_list`` is serialized by ``_update_tool_list_lock`` (no interleaving)
+- The shared ``servers`` cross-request cache evicts oldest entries when it exceeds
+  ``SHARED_SERVERS_CACHE_MAX_ENTRIES``
+- The Toolset output is declared / persisted with ``cache=False`` so saved flows
+  do not memoize a stale tool list
+"""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from lfx.base.agents.utils import safe_cache_get, safe_cache_set
+from lfx.base.tools.constants import TOOL_OUTPUT_NAME
+from lfx.components.models_and_agents.mcp_component import MCPToolsComponent
+
+
+def _make_tool(name: str) -> MagicMock:
+    tool = MagicMock()
+    tool.name = name
+    return tool
+
+
+class TestHeaderNormalization:
+    """``_normalized_headers_for_cache`` is the input to the cache-key hash."""
+
+    def test_list_of_key_value_dicts(self) -> None:
+        component = MCPToolsComponent()
+        component.headers = [
+            {"key": "Authorization", "value": "Bearer abc"},
+            {"key": "X-Tenant", "value": "acme"},
+        ]
+
+        assert component._normalized_headers_for_cache() == {
+            "Authorization": "Bearer abc",
+            "X-Tenant": "acme",
+        }
+
+    def test_dict_input(self) -> None:
+        component = MCPToolsComponent()
+        component.headers = {"Authorization": "Bearer abc"}
+
+        assert component._normalized_headers_for_cache() == {"Authorization": "Bearer abc"}
+
+    def test_none_returns_empty_dict(self) -> None:
+        component = MCPToolsComponent()
+        component.headers = None
+
+        assert component._normalized_headers_for_cache() == {}
+
+    def test_malformed_list_items_are_skipped(self) -> None:
+        component = MCPToolsComponent()
+        component.headers = [
+            {"key": "Authorization", "value": "Bearer abc"},
+            "not-a-dict",
+            {"no_key": "bad"},
+        ]
+
+        assert component._normalized_headers_for_cache() == {"Authorization": "Bearer abc"}
+
+
+class TestCacheKey:
+    """``_mcp_servers_cache_key`` must separate auth contexts and stay deterministic."""
+
+    def test_empty_server_name_returns_empty_string(self) -> None:
+        component = MCPToolsComponent()
+        assert component._mcp_servers_cache_key("") == ""
+
+    def test_no_headers_returns_bare_server_name(self) -> None:
+        component = MCPToolsComponent()
+        component.headers = []
+
+        assert component._mcp_servers_cache_key("srv") == "srv"
+
+    def test_different_headers_produce_different_keys(self) -> None:
+        a = MCPToolsComponent()
+        a.headers = [{"key": "Authorization", "value": "Bearer tenant-a"}]
+        b = MCPToolsComponent()
+        b.headers = [{"key": "Authorization", "value": "Bearer tenant-b"}]
+
+        assert a._mcp_servers_cache_key("srv") != b._mcp_servers_cache_key("srv")
+
+    def test_same_headers_produce_identical_keys(self) -> None:
+        a = MCPToolsComponent()
+        a.headers = [{"key": "Authorization", "value": "Bearer same"}]
+        b = MCPToolsComponent()
+        b.headers = [{"key": "Authorization", "value": "Bearer same"}]
+
+        assert a._mcp_servers_cache_key("srv") == b._mcp_servers_cache_key("srv")
+
+    def test_header_order_does_not_change_key(self) -> None:
+        a = MCPToolsComponent()
+        a.headers = [
+            {"key": "Authorization", "value": "Bearer x"},
+            {"key": "X-Tenant", "value": "acme"},
+        ]
+        b = MCPToolsComponent()
+        b.headers = [
+            {"key": "X-Tenant", "value": "acme"},
+            {"key": "Authorization", "value": "Bearer x"},
+        ]
+
+        assert a._mcp_servers_cache_key("srv") == b._mcp_servers_cache_key("srv")
+
+
+class TestTtlToolCacheIsolation:
+    """``_ttl_tool_cache`` must be a per-instance dict, not class-level."""
+
+    def test_fresh_instances_have_independent_dicts(self) -> None:
+        a = MCPToolsComponent()
+        b = MCPToolsComponent()
+
+        assert a._ttl_tool_cache is not b._ttl_tool_cache
+
+    def test_write_to_one_instance_does_not_leak_to_another(self) -> None:
+        a = MCPToolsComponent()
+        b = MCPToolsComponent()
+        a._ttl_tool_cache["k"] = (0.0, [_make_tool("leak")])
+
+        assert "k" not in b._ttl_tool_cache
+
+
+class TestGetToolsTtlCache:
+    """``_get_tools`` uses the per-instance TTL cache with FIFO eviction + expiry."""
+
+    @pytest.mark.asyncio
+    async def test_ttl_cache_hit_skips_update_tool_list(self) -> None:
+        component = MCPToolsComponent()
+        component.mcp_server = {"name": "srv"}
+        component.headers = []
+
+        cached_tools = [_make_tool("cached")]
+        ttl_key = component._mcp_servers_cache_key("srv")
+        import time as _time
+
+        component._ttl_tool_cache[ttl_key] = (_time.monotonic(), cached_tools)
+
+        with patch.object(component, "update_tool_list", new=AsyncMock()) as mocked_update:
+            result = await component._get_tools()
+
+        assert result is cached_tools
+        mocked_update.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_ttl_cache_expired_entry_is_refetched(self) -> None:
+        component = MCPToolsComponent()
+        component.mcp_server = {"name": "srv"}
+        component.headers = []
+
+        stale_tools = [_make_tool("stale")]
+        fresh_tools = [_make_tool("fresh")]
+        ttl_key = component._mcp_servers_cache_key("srv")
+        # Timestamp older than TOOL_TTL_SECS so the entry is considered expired.
+        component._ttl_tool_cache[ttl_key] = (0.0, stale_tools)
+
+        with patch.object(
+            component,
+            "update_tool_list",
+            new=AsyncMock(return_value=(fresh_tools, {"name": "srv", "config": {}})),
+        ):
+            result = await component._get_tools()
+
+        assert result is fresh_tools
+        # Stale entry was evicted and replaced with the fresh one.
+        assert component._ttl_tool_cache[ttl_key][1] is fresh_tools
+
+    @pytest.mark.asyncio
+    async def test_ttl_cache_bounded_by_max_entries_fifo(self) -> None:
+        component = MCPToolsComponent()
+        # Shrink the cap for a fast, deterministic FIFO check.
+        component.TOOL_TTL_MAX_ENTRIES = 3
+
+        async def fake_update_tool_list(mcp_server):
+            srv = mcp_server.get("name") if isinstance(mcp_server, dict) else mcp_server
+            return [_make_tool(f"{srv}-tool")], {"name": srv, "config": {}}
+
+        with patch.object(component, "update_tool_list", new=AsyncMock(side_effect=fake_update_tool_list)):
+            for i in range(5):
+                component.mcp_server = {"name": f"srv-{i}"}
+                component.headers = []
+                await component._get_tools()
+
+        assert len(component._ttl_tool_cache) == 3
+        # FIFO eviction: the first two inserted keys must be gone, the last three must remain.
+        remaining_keys = set(component._ttl_tool_cache.keys())
+        for i in (0, 1):
+            assert component._mcp_servers_cache_key(f"srv-{i}") not in remaining_keys
+        for i in (2, 3, 4):
+            assert component._mcp_servers_cache_key(f"srv-{i}") in remaining_keys
+
+    @pytest.mark.asyncio
+    async def test_ttl_cache_disabled_when_ttl_is_zero(self) -> None:
+        component = MCPToolsComponent()
+        component.TOOL_TTL_SECS = 0
+        component.mcp_server = {"name": "srv"}
+        component.headers = []
+
+        fresh_tools = [_make_tool("fresh")]
+        with patch.object(
+            component,
+            "update_tool_list",
+            new=AsyncMock(return_value=(fresh_tools, {"name": "srv", "config": {}})),
+        ):
+            await component._get_tools()
+
+        # With TTL disabled, nothing should ever be written to the cache.
+        assert component._ttl_tool_cache == {}
+
+
+class TestUpdateToolListLock:
+    """Concurrent ``update_tool_list`` calls must be serialized per component."""
+
+    @pytest.mark.asyncio
+    async def test_concurrent_calls_are_serialized(self) -> None:
+        component = MCPToolsComponent()
+        component.use_cache = False
+
+        in_flight = 0
+        peak = 0
+        entered = asyncio.Event()
+
+        async def stub_run(_mcp_server):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            entered.set()
+            await asyncio.sleep(0.02)
+            in_flight -= 1
+            return [], {"name": "srv", "config": {}}
+
+        async def guarded(mcp_server):
+            async with component._update_tool_list_lock:
+                return await stub_run(mcp_server)
+
+        # Ten concurrent invocations must not overlap because the lock serializes them.
+        await asyncio.gather(*(guarded("srv") for _ in range(10)))
+        assert peak == 1
+        assert entered.is_set()
+
+
+class TestSharedServersCacheEviction:
+    """Shared ``servers`` cache must be bounded by ``SHARED_SERVERS_CACHE_MAX_ENTRIES``."""
+
+    @pytest.mark.asyncio
+    async def test_fifo_eviction_when_over_limit(self) -> None:
+        component = MCPToolsComponent()
+        component.SHARED_SERVERS_CACHE_MAX_ENTRIES = 3
+
+        # Seed the shared cache up to capacity with placeholder entries.
+        servers_cache: dict = {}
+        for i in range(3):
+            servers_cache[f"old-{i}"] = {
+                "tools": [],
+                "tool_names": [],
+                "tool_cache": {},
+                "config": {"i": i},
+            }
+        safe_cache_set(component._shared_component_cache, "servers", servers_cache)
+
+        # The component's update_tool_list block applies FIFO eviction before inserting a new
+        # key whenever len >= max_entries and the new key is absent. Reproduce that block here
+        # to exercise the exact policy the component uses.
+        new_key = "new-key"
+        cache_data = {
+            "tools": [],
+            "tool_names": [],
+            "tool_cache": {},
+            "config": {"new": True},
+        }
+        current = safe_cache_get(component._shared_component_cache, "servers", {})
+        max_entries = component.SHARED_SERVERS_CACHE_MAX_ENTRIES
+        while len(current) >= max_entries and new_key not in current:
+            oldest = next(iter(current))
+            current.pop(oldest, None)
+        current[new_key] = cache_data
+        safe_cache_set(component._shared_component_cache, "servers", current)
+
+        final = safe_cache_get(component._shared_component_cache, "servers", {})
+        assert len(final) == 3
+        assert new_key in final
+        # The oldest entry ("old-0") must have been evicted first.
+        assert "old-0" not in final
+        assert "old-1" in final
+        assert "old-2" in final
+
+
+class TestToolsetOutputNotCached:
+    """Saved flows must not memoize the Toolset output."""
+
+    def test_build_tool_output_declares_cache_false(self) -> None:
+        component = MCPToolsComponent()
+        output = component._build_tool_output()
+
+        assert output.name == TOOL_OUTPUT_NAME
+        assert output.cache is False
+
+    def test_map_outputs_forces_cache_false_on_persisted_output(self) -> None:
+        component = MCPToolsComponent()
+        # Seed the outputs map with an entry that *claims* cache=True, as saved flow JSON
+        # occasionally does. ``map_outputs`` must override it back to False.
+        persisted = MagicMock()
+        persisted.cache = True
+        component._outputs_map = {TOOL_OUTPUT_NAME: persisted}
+
+        # Short-circuit the super() call; this test isolates the override behaviour.
+        with patch(
+            "lfx.custom.custom_component.component_with_cache.ComponentWithCache.map_outputs",
+            return_value=None,
+        ):
+            component.map_outputs()
+
+        assert component._outputs_map[TOOL_OUTPUT_NAME].cache is False

src/lfx/src/lfx/components/models_and_agents/mcp_component.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import json
+import time
 import uuid
 from types import UnionType
 from typing import Any, get_args, get_origin
@@ -15,6 +17,7 @@ from lfx.base.mcp.util import (
     MCPStreamableHttpClient,
     update_tools,
 )
+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
@@ -51,6 +54,37 @@ def resolve_mcp_config(
 
 
 class MCPToolsComponent(ComponentWithCache):
+    """MCP Tools component.
+
+    Behaviour notes:
+    - Stale agent tools vs server were caused by Langflow caching the Toolset output
+      (``Output.cache=True`` plus the persisted ``output.value`` in saved flow JSON), not
+      only by the user-facing "Use Cached Server" (``use_cache``) toggle. ``_build_tool_output``
+      declares the Toolset output with ``cache=False`` and ``map_outputs`` overrides the
+      persisted value for existing flows so every run resolves a fresh tool list.
+    - ``update_tool_list`` is serialized with an ``asyncio.Lock`` so concurrent calls do not
+      share the same Streamable HTTP client session; overlapping POST/DELETE cycles otherwise
+      surface as HTTP 404 and the MCP SDK reports ``Session terminated``.
+    - All diagnostic logs are at DEBUG level to avoid swamping logs on hot paths (an agent
+      may call ``_get_tools`` per step). Header values are never logged; only header *keys*.
+    """
+
+    # Short-lived in-memory cache window (seconds) for ``_get_tools``. When the
+    # same header-hash key is requested again within this window the cached tool
+    # list is reused so parallel agent runs that share identical auth don't
+    # each pay for a fresh MCP round-trip. Set to ``0`` to disable. This is
+    # distinct from the "Use Cached Server" (``use_cache``) toggle which only
+    # controls the shared cross-request server cache.
+    TOOL_TTL_SECS: int = 30
+    # Upper bound on the per-instance TTL cache. Oldest entries are evicted
+    # when the cap is reached. Keeps memory bounded for flows where the same
+    # component handles many rotating auth contexts (e.g. per-tenant tokens).
+    TOOL_TTL_MAX_ENTRIES: int = 32
+    # Upper bound on the shared cross-request ``servers`` cache. Each
+    # (server_name, header-hash) pair is a distinct entry; without a bound a
+    # tenant that rotates session tokens would grow this map without limit.
+    SHARED_SERVERS_CACHE_MAX_ENTRIES: int = 64
+
     schema_inputs: list = []
     tools: list[StructuredTool] = []
     _not_load_actions: bool = False
@@ -67,6 +101,14 @@ class MCPToolsComponent(ComponentWithCache):
         self.streamable_http_client: MCPStreamableHttpClient = MCPStreamableHttpClient(
             component_cache=self._shared_component_cache
         )
+        # One MCP stdio/streamable client pair per component; concurrent update_tool_list calls
+        # otherwise race (session DELETE vs POST) and the MCP SDK surfaces HTTP 404 as "Session terminated".
+        self._update_tool_list_lock = asyncio.Lock()
+        # Per-instance TTL cache for ``_get_tools``: {cache_key: (monotonic_ts, tools)}.
+        # Declared here (not at class scope) so every component gets its own dict —
+        # a class-level dict would be shared across every MCPToolsComponent in the process,
+        # leaking tool lists across tenants that happen to hash to the same key.
+        self._ttl_tool_cache: dict[str, tuple[float, list]] = {}
 
     def _ensure_cache_structure(self):
         """Ensure the cache has the required structure."""
@@ -80,6 +122,55 @@ class MCPToolsComponent(ComponentWithCache):
         if last_server_value is None:
             safe_cache_set(self._shared_component_cache, "last_selected_server", "")
 
+    def _normalized_headers_for_cache(self) -> dict[str, str]:
+        """Component headers as a dict for stable cache keying (auth / tweaks)."""
+        component_headers = getattr(self, "headers", None) or []
+        if isinstance(component_headers, list):
+            return {
+                str(item["key"]): str(item["value"])
+                for item in component_headers
+                if isinstance(item, dict) and "key" in item and "value" in item
+            }
+        if isinstance(component_headers, dict):
+            return {str(k): str(v) for k, v in component_headers.items()}
+        return {}
+
+    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."""
+        if not server_name:
+            return ""
+        hdrs = self._normalized_headers_for_cache()
+        if not hdrs:
+            return server_name
+        payload = json.dumps(hdrs, sort_keys=True)
+        digest = hashlib.sha256(payload.encode()).hexdigest()[:16]
+        return f"{server_name}:{digest}"
+
+    def _build_tool_output(self) -> Output:
+        # Do not cache Toolset output. This is separate from the MCP "Use Cached Server" (use_cache)
+        # toggle: Langflow's Output.cache defaults to True and was memoizing the first to_toolkit()
+        # result, so per-request tweaks/headers never refreshed bound tools even when use_cache=False.
+        return Output(
+            name=TOOL_OUTPUT_NAME,
+            display_name=TOOL_OUTPUT_DISPLAY_NAME,
+            method="to_toolkit",
+            types=["Tool"],
+            cache=False,
+        )
+
+    def map_outputs(self) -> None:
+        """Override the persisted ``component_as_tool`` cache flag from saved flow JSON.
+
+        ``_build_tool_output`` already returns the output with ``cache=False``, but the
+        flow JSON for existing flows often stores ``cache: true`` for ``component_as_tool``
+        and that persisted value wins over the declaration. Forcing ``cache=False`` here
+        guarantees saved flows also bypass Output memoization and get a fresh tool list
+        on every run.
+        """
+        super().map_outputs()
+        if TOOL_OUTPUT_NAME in self._outputs_map:
+            self._outputs_map[TOOL_OUTPUT_NAME].cache = False
+
     default_keys: list[str] = [
         "code",
         "_type",
@@ -198,164 +289,218 @@ class MCPToolsComponent(ComponentWithCache):
             server_name = mcp_server
         if not server_name:
             self.tools = []
+            await logger.adebug("MCP update_tool_list: empty server_name, clearing tools")
             return [], {"name": server_name, "config": server_config_from_value}
 
+        servers_cache_key = self._mcp_servers_cache_key(server_name)
+
         # Check if caching is enabled, default to False
         use_cache = getattr(self, "use_cache", False)
+        header_keys = sorted(self._normalized_headers_for_cache().keys())
+        await logger.adebug(
+            "MCP update_tool_list: start server=%r use_cache=%s shared_cache_key=%r header_keys=%s",
+            server_name,
+            use_cache,
+            servers_cache_key,
+            header_keys,
+        )
 
-        # Use shared cache if available and caching is enabled
-        cached = None
-        if use_cache:
-            servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
-            cached = servers_cache.get(server_name) if isinstance(servers_cache, dict) else None
-
-        if cached is not None:
-            try:
-                self.tools = cached["tools"]
-                self.tool_names = cached["tool_names"]
-                self._tool_cache = cached["tool_cache"]
-                server_config_from_value = cached["config"]
-            except (TypeError, KeyError, AttributeError) as e:
-                # Handle corrupted cache data by clearing it and continuing to fetch fresh tools
-                msg = f"Unable to use cached data for MCP Server{server_name}: {e}"
-                await logger.awarning(msg)
-                # Clear the corrupted cache entry
-                current_servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
-                if isinstance(current_servers_cache, dict) and server_name in current_servers_cache:
-                    current_servers_cache.pop(server_name)
-                    safe_cache_set(self._shared_component_cache, "servers", current_servers_cache)
-            else:
-                return self.tools, {"name": server_name, "config": server_config_from_value}
-
-        try:
-            # Try to fetch from database first to ensure we have the latest config
-            # This ensures database updates (like editing a server) take effect
-            try:
-                from langflow.api.v2.mcp import get_server
-                from langflow.services.database.models.user.crud import get_user_by_id
-
-                from lfx.services.deps import get_settings_service
-            except ImportError as e:
-                msg = (
-                    "Langflow MCP server functionality is not available. "
-                    "This feature requires the full Langflow installation."
-                )
-                raise ImportError(msg) from e
-
-            server_config_from_db = None
-            async with session_scope() as db:
-                if not self.user_id:
-                    msg = "User ID is required for fetching MCP tools."
-                    raise ValueError(msg)
-                current_user = await get_user_by_id(db, self.user_id)
-
-                # Try to get server config from DB/API
-                server_config_from_db = await get_server(
-                    server_name,
-                    current_user,
-                    db,
-                    storage_service=get_storage_service(),
-                    settings_service=get_settings_service(),
-                )
-
-            # Resolve config with proper precedence: DB takes priority, falls back to value
-            server_config = resolve_mcp_config(
-                server_name=server_name,
-                server_config_from_value=server_config_from_value,
-                server_config_from_db=server_config_from_db,
-            )
-
-            if not server_config:
-                self.tools = []
-                return [], {"name": server_name, "config": server_config}
-
-            # Add verify_ssl option to server config if not present
-            if "verify_ssl" not in server_config:
-                verify_ssl = getattr(self, "verify_ssl", True)
-                server_config["verify_ssl"] = verify_ssl
-
-            # Merge headers from component input with server config headers
-            # Component headers take precedence over server config headers
-            component_headers = getattr(self, "headers", None) or []
-            if component_headers:
-                # Convert list of {"key": k, "value": v} to dict
-                component_headers_dict = {}
-                if isinstance(component_headers, list):
-                    for item in component_headers:
-                        if isinstance(item, dict) and "key" in item and "value" in item:
-                            component_headers_dict[item["key"]] = item["value"]
-                elif isinstance(component_headers, dict):
-                    component_headers_dict = component_headers
-
-                if component_headers_dict:
-                    existing_headers = server_config.get("headers", {}) or {}
-                    # Ensure existing_headers is a dict (convert from list if needed)
-                    if isinstance(existing_headers, list):
-                        existing_dict = {}
-                        for item in existing_headers:
-                            if isinstance(item, dict) and "key" in item and "value" in item:
-                                existing_dict[item["key"]] = item["value"]
-                        existing_headers = existing_dict
-                    merged_headers = {**existing_headers, **component_headers_dict}
-                    server_config["headers"] = merged_headers
-            # Get request_variables from graph context for global variable resolution
-            request_variables = None
-            if hasattr(self, "graph") and self.graph and hasattr(self.graph, "context"):
-                request_variables = self.graph.context.get("request_variables")
-
-            # Only load global variables from database if we have headers that might use them
-            # This avoids unnecessary database queries when headers are empty
-            has_headers = server_config.get("headers") and len(server_config.get("headers", {})) > 0
-            if not request_variables and has_headers:
-                try:
-                    from lfx.services.deps import get_variable_service
-
-                    variable_service = get_variable_service()
-                    if variable_service:
-                        async with session_scope() as db:
-                            request_variables = await variable_service.get_all_decrypted_variables(
-                                user_id=self.user_id, session=db
-                            )
-                except Exception as e:  # noqa: BLE001
-                    await logger.awarning(f"Failed to load global variables for MCP component: {e}")
-
-            _, 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,
-            )
-
-            self.tool_names = [tool.name for tool in tool_list if hasattr(tool, "name")]
-            self._tool_cache = tool_cache
-            self.tools = tool_list
-
-            # Cache the result only if caching is enabled
+        async with self._update_tool_list_lock:
+            # Use shared cache if available and caching is enabled
+            cached = None
             if use_cache:
-                cache_data = {
-                    "tools": tool_list,
-                    "tool_names": self.tool_names,
-                    "tool_cache": tool_cache,
-                    "config": server_config,
-                }
+                servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
+                cached = servers_cache.get(servers_cache_key) if isinstance(servers_cache, dict) else None
 
-                # Safely update the servers cache
-                current_servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
-                if isinstance(current_servers_cache, dict):
-                    current_servers_cache[server_name] = cache_data
-                    safe_cache_set(self._shared_component_cache, "servers", current_servers_cache)
+            if cached is not None:
+                try:
+                    tools_from_cache = cached["tools"]
+                    server_config_from_value = cached["config"]
+                except (TypeError, KeyError, AttributeError) as e:
+                    # Handle corrupted cache data by clearing it and continuing to fetch fresh tools
+                    msg = f"Unable to use cached data for MCP Server{server_name}: {e}"
+                    await logger.awarning(msg)
+                    # Clear the corrupted cache entry
+                    current_servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
+                    if isinstance(current_servers_cache, dict) and servers_cache_key in current_servers_cache:
+                        current_servers_cache.pop(servers_cache_key)
+                        safe_cache_set(self._shared_component_cache, "servers", current_servers_cache)
+                else:
+                    self.tools = tools_from_cache
+                    self.tool_names = [t.name for t in self.tools if hasattr(t, "name")]
+                    self._tool_cache = cached["tool_cache"]
+                    await logger.adebug(
+                        "MCP update_tool_list: shared_servers_cache HIT count=%d server=%r",
+                        len(self.tools),
+                        server_name,
+                    )
+                    return self.tools, {"name": server_name, "config": server_config_from_value}
 
-        except (TimeoutError, asyncio.TimeoutError) as e:
-            msg = f"Timeout updating tool list: {e!s}"
-            await logger.aexception(msg)
-            raise TimeoutError(msg) from e
-        except Exception as e:
-            msg = f"Error updating tool list: {e!s}"
-            await logger.aexception(msg)
-            raise ValueError(msg) from e
-        else:
-            return tool_list, {"name": server_name, "config": server_config}
+            try:
+                # Try to fetch from database first to ensure we have the latest config
+                # This ensures database updates (like editing a server) take effect
+                try:
+                    from langflow.api.v2.mcp import get_server
+                    from langflow.services.database.models.user.crud import get_user_by_id
+
+                    from lfx.services.deps import get_settings_service
+                except ImportError as e:
+                    msg = (
+                        "Langflow MCP server functionality is not available. "
+                        "This feature requires the full Langflow installation."
+                    )
+                    raise ImportError(msg) from e
+
+                server_config_from_db = None
+                async with session_scope() as db:
+                    if not self.user_id:
+                        msg = "User ID is required for fetching MCP tools."
+                        raise ValueError(msg)
+                    current_user = await get_user_by_id(db, self.user_id)
+
+                    # Try to get server config from DB/API
+                    server_config_from_db = await get_server(
+                        server_name,
+                        current_user,
+                        db,
+                        storage_service=get_storage_service(),
+                        settings_service=get_settings_service(),
+                    )
+
+                # Resolve config with proper precedence: DB takes priority, falls back to value
+                server_config = resolve_mcp_config(
+                    server_name=server_name,
+                    server_config_from_value=server_config_from_value,
+                    server_config_from_db=server_config_from_db,
+                )
+
+                if not server_config:
+                    self.tools = []
+                    await logger.awarning(
+                        "MCP update_tool_list: no server_config after resolve server=%r",
+                        server_name,
+                    )
+                    return [], {"name": server_name, "config": server_config}
+
+                # Add verify_ssl option to server config if not present
+                if "verify_ssl" not in server_config:
+                    verify_ssl = getattr(self, "verify_ssl", True)
+                    server_config["verify_ssl"] = verify_ssl
+
+                # Merge headers from component input with server config headers
+                # Component headers take precedence over server config headers
+                component_headers = getattr(self, "headers", None) or []
+                if component_headers:
+                    # Convert list of {"key": k, "value": v} to dict
+                    component_headers_dict = {}
+                    if isinstance(component_headers, list):
+                        for item in component_headers:
+                            if isinstance(item, dict) and "key" in item and "value" in item:
+                                component_headers_dict[item["key"]] = item["value"]
+                    elif isinstance(component_headers, dict):
+                        component_headers_dict = component_headers
+
+                    if component_headers_dict:
+                        existing_headers = server_config.get("headers", {}) or {}
+                        # Ensure existing_headers is a dict (convert from list if needed)
+                        if isinstance(existing_headers, list):
+                            existing_dict = {}
+                            for item in existing_headers:
+                                if isinstance(item, dict) and "key" in item and "value" in item:
+                                    existing_dict[item["key"]] = item["value"]
+                            existing_headers = existing_dict
+                        merged_headers = {**existing_headers, **component_headers_dict}
+                        server_config["headers"] = merged_headers
+                # Get request_variables from graph context for global variable resolution
+                request_variables = None
+                if hasattr(self, "graph") and self.graph and hasattr(self.graph, "context"):
+                    request_variables = self.graph.context.get("request_variables")
+
+                # Only load global variables from database if we have headers that might use them
+                # This avoids unnecessary database queries when headers are empty
+                has_headers = server_config.get("headers") and len(server_config.get("headers", {})) > 0
+                if not request_variables and has_headers:
+                    try:
+                        from lfx.services.deps import get_variable_service
+
+                        variable_service = get_variable_service()
+                        if variable_service:
+                            async with session_scope() as db:
+                                request_variables = await variable_service.get_all_decrypted_variables(
+                                    user_id=self.user_id, session=db
+                                )
+                    except Exception as e:  # noqa: BLE001
+                        await logger.awarning(f"Failed to load global variables for MCP component: {e}")
+
+                await logger.adebug(
+                    "MCP update_tool_list: calling update_tools server=%r mode_headers=%s",
+                    server_name,
+                    sorted((server_config.get("headers") or {}).keys())
+                    if isinstance(server_config.get("headers"), dict)
+                    else "list-or-empty",
+                )
+
+                _, 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,
+                )
+
+                self.tool_names = [tool.name for tool in tool_list if hasattr(tool, "name")]
+                self._tool_cache = tool_cache
+                self.tools = tool_list
+
+                await logger.adebug(
+                    "MCP update_tool_list: fetched from MCP count=%d server=%r",
+                    len(tool_list),
+                    server_name,
+                )
+
+                # Cache the result only if caching is enabled
+                if use_cache:
+                    cache_data = {
+                        "tools": tool_list,
+                        "tool_names": self.tool_names,
+                        "tool_cache": tool_cache,
+                        "config": server_config,
+                    }
+
+                    # Safely update the servers cache with bounded size (FIFO eviction).
+                    current_servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
+                    if isinstance(current_servers_cache, dict):
+                        # Because the cache key now includes a header hash, a tenant that
+                        # rotates session tokens would grow this map without bound. Drop
+                        # the oldest entry when over the limit.
+                        max_entries = self.SHARED_SERVERS_CACHE_MAX_ENTRIES
+                        while (
+                            len(current_servers_cache) >= max_entries and servers_cache_key not in current_servers_cache
+                        ):
+                            oldest_key = next(iter(current_servers_cache))
+                            current_servers_cache.pop(oldest_key, None)
+                        current_servers_cache[servers_cache_key] = cache_data
+                        safe_cache_set(self._shared_component_cache, "servers", current_servers_cache)
+                        await logger.adebug(
+                            "MCP update_tool_list: wrote shared_servers_cache key=%r size=%d",
+                            servers_cache_key,
+                            len(current_servers_cache),
+                        )
+
+            except (TimeoutError, asyncio.TimeoutError) as e:
+                msg = (
+                    f"Timeout updating tool list: {e!s}. "
+                    "Raise ``LANGFLOW_MCP_SERVER_TIMEOUT`` for the deployment if the MCP "
+                    "server legitimately needs more time to respond."
+                )
+                await logger.aexception(msg)
+                raise TimeoutError(msg) from e
+            except Exception as e:
+                msg = f"Error updating tool list: {e!s}"
+                await logger.aexception(msg)
+                raise ValueError(msg) from e
+            else:
+                return tool_list, {"name": server_name, "config": server_config}
 
     async def update_build_config(self, build_config: dict, field_value: str, field_name: str | None = None) -> dict:
         """Toggle the visibility of connection-specific fields based on the selected mode."""
@@ -420,6 +565,7 @@ class MCPToolsComponent(ComponentWithCache):
                 build_config["tool_placeholder"]["tool_mode"] = True
 
                 current_server_name = field_value.get("name") if isinstance(field_value, dict) else field_value
+                servers_cache_key_ui = self._mcp_servers_cache_key(current_server_name) if current_server_name else ""
                 _last_selected_server = safe_cache_get(self._shared_component_cache, "last_selected_server", "")
                 # Only treat as a server change if there was a previous server selection.
                 # Cold cache (_last_selected_server="") on initial flow load is NOT a server change —
@@ -452,7 +598,7 @@ class MCPToolsComponent(ComponentWithCache):
                     if current_server_name:
                         servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
                         if isinstance(servers_cache, dict):
-                            cached = servers_cache.get(current_server_name)
+                            cached = servers_cache.get(servers_cache_key_ui)
                             if cached is not None and cached.get("tool_names"):
                                 cached_tools = cached["tool_names"]
                                 current_tools = build_config["tool"]["options"]
@@ -466,8 +612,8 @@ class MCPToolsComponent(ComponentWithCache):
                 # This ensures we always fetch fresh data from the database
                 if not use_cache and current_server_name:
                     servers_cache = safe_cache_get(self._shared_component_cache, "servers", {})
-                    if isinstance(servers_cache, dict) and current_server_name in servers_cache:
-                        servers_cache.pop(current_server_name)
+                    if isinstance(servers_cache, dict) and servers_cache_key_ui in servers_cache:
+                        servers_cache.pop(servers_cache_key_ui)
                         safe_cache_set(self._shared_component_cache, "servers", servers_cache)
 
                 # Check if tools are already cached for this server before clearing
@@ -774,9 +920,50 @@ class MCPToolsComponent(ComponentWithCache):
         return None
 
     async def _get_tools(self):
-        """Get cached tools or update if necessary."""
+        """Load tools for the agent toolkit; always refresh from ``update_tool_list``.
+
+        ``_not_load_actions`` only applies to UI build_config flows (tool dropdown). Agent
+        runs must always bind the current tool list (including after header/tweak changes).
+
+        A short-lived TTL cache (``TOOL_TTL_SECS``, header-hash-keyed) skips the MCP
+        round-trip when the same auth context is re-queried quickly (e.g. parallel agent
+        steps sharing the same tweaked headers). The cache is per component instance and
+        bounded by ``TOOL_TTL_MAX_ENTRIES``; it is distinct from the "Use Cached Server"
+        (``use_cache``) toggle which controls the shared cross-request cache.
+        """
         mcp_server = getattr(self, "mcp_server", None)
-        if not self._not_load_actions:
-            tools, _ = await self.update_tool_list(mcp_server)
-            return tools
-        return []
+        srv = mcp_server.get("name") if isinstance(mcp_server, dict) else mcp_server
+
+        ttl = self.TOOL_TTL_SECS
+        ttl_key = self._mcp_servers_cache_key(srv or "") if ttl > 0 and srv else ""
+
+        # TTL cache lookup + expired-entry eviction.
+        if ttl > 0 and ttl_key:
+            cached = self._ttl_tool_cache.get(ttl_key)
+            if cached is not None:
+                ts, cached_tools = cached
+                age = time.monotonic() - ts
+                if age < ttl:
+                    await logger.adebug(
+                        "MCP _get_tools: TTL cache hit count=%d age=%.1fs server=%r",
+                        len(cached_tools),
+                        age,
+                        srv,
+                    )
+                    return cached_tools
+                # Stale — drop it so the size check below stays tight.
+                self._ttl_tool_cache.pop(ttl_key, None)
+
+        await logger.adebug("MCP _get_tools: fetching tool list server=%r", srv)
+        tools, _ = await self.update_tool_list(mcp_server)
+        await logger.adebug("MCP _get_tools: fetched %d tools server=%r", len(tools), srv)
+
+        # TTL cache store with bounded size (FIFO eviction of the oldest entry).
+        if ttl > 0 and ttl_key and tools:
+            if len(self._ttl_tool_cache) >= self.TOOL_TTL_MAX_ENTRIES:
+                # dict preserves insertion order — pop the oldest entry.
+                oldest_key = next(iter(self._ttl_tool_cache))
+                self._ttl_tool_cache.pop(oldest_key, None)
+            self._ttl_tool_cache[ttl_key] = (time.monotonic(), tools)
+
+        return tools