fix(mcp): reconcile live agent tool snapshots

openhands-agent-server/openhands/agent_server/mcp_oauth_store.py

@@ -24,7 +24,10 @@
     MCPOAuthTokenStorageField,
     MCPServer,
 )
-from openhands.sdk.mcp.utils import ToolsChangedCallback, create_mcp_tools
+from openhands.sdk.mcp.utils import (
+    ToolsChangedCallback,
+    create_mcp_tools,
+)
 
 
 logger = get_logger(__name__)

openhands-sdk/openhands/sdk/agent/base.py

@@ -3,6 +3,7 @@
 import os
 import re
 import sys
+import threading
 from abc import ABC, abstractmethod
 from collections import Counter
 from collections.abc import Generator, Iterable, Sequence
@@ -27,8 +28,9 @@
 from openhands.sdk.llm import LLM
 from openhands.sdk.llm.utils.model_prompt_spec import get_model_prompt_spec
 from openhands.sdk.logger import get_logger
+from openhands.sdk.mcp.client import MCPClient
 from openhands.sdk.mcp.config import MCPServer
-from openhands.sdk.mcp.tool import MCPToolExecutor
+from openhands.sdk.mcp.tool import MCPToolDefinition, MCPToolExecutor
 from openhands.sdk.tool import (
     BUILT_IN_TOOL_CLASSES,
     BUILT_IN_TOOLS,
@@ -300,6 +302,7 @@ def _validate_system_prompt_fields(cls, data: Any) -> Any:
 
     # Runtime materialized tools; private and non-serializable
     _tools: dict[str, ToolDefinition] = PrivateAttr(default_factory=dict)
+    _tools_lock: threading.RLock = PrivateAttr(default_factory=threading.RLock)
     _initialized: bool = PrivateAttr(default=False)
 
     @property
@@ -561,6 +564,7 @@ def _initialize(
         if self.filter_tools_regex:
             pattern = re.compile(self.filter_tools_regex)
             tools = [tool for tool in tools if pattern.match(tool.name)]
+            tool_names = [tool.name for tool in tools]
             logger.info("Filtered to %d tools after applying regex filter", len(tools))
 
         # Include default tools from include_default_tools; not subject to regex
@@ -867,13 +871,14 @@ def add_runtime_tools(self, tools: Sequence[ToolDefinition]) -> None:
                 name for name, count in Counter(tool_names).items() if count > 1
             }
             raise ValueError(f"Duplicate runtime tool names found: {duplicates}")
-        existing = set(self._tools) & set(tool_names)
-        if existing:
-            raise ValueError(f"Duplicate tool names found: {existing}")
+        with self._tools_lock:
+            existing = set(self._tools) & set(tool_names)
+            if existing:
+                raise ValueError(f"Duplicate tool names found: {existing}")
 
-        # AgentBase is frozen, so update its mutable tool map in place.
-        for tool in tools:
-            self._tools[tool.name] = tool
+            # AgentBase is frozen, so update its mutable tool map in place.
+            for tool in tools:
+                self._tools[tool.name] = tool
 
     def _on_mcp_tools_changed(self, tools: Sequence[ToolDefinition]) -> None:
         """Handle dynamically advertised MCP tools.
@@ -897,35 +902,36 @@ def _on_mcp_tools_changed(self, tools: Sequence[ToolDefinition]) -> None:
             }
             raise ValueError(f"Duplicate MCP tool names found: {duplicates}")
 
-        additions: list[ToolDefinition] = []
-        replacements: list[ToolDefinition] = []
-        conflicts: set[str] = set()
-        for tool in tools:
-            existing = self._tools.get(tool.name)
-            if existing is None:
-                additions.append(tool)
-                continue
-
-            existing_executor = existing.executor
-            replacement_executor = tool.executor
-            if (
-                isinstance(existing_executor, MCPToolExecutor)
-                and isinstance(replacement_executor, MCPToolExecutor)
-                and existing_executor.client is replacement_executor.client
-            ):
-                replacements.append(tool)
-            else:
-                conflicts.add(tool.name)
-
-        if conflicts:
-            raise ValueError(
-                "Dynamically advertised MCP tools conflict with existing runtime "
-                f"tools: {sorted(conflicts)}"
-            )
+        with self._tools_lock:
+            additions: list[ToolDefinition] = []
+            replacements: list[ToolDefinition] = []
+            conflicts: set[str] = set()
+            for tool in tools:
+                existing = self._tools.get(tool.name)
+                if existing is None:
+                    additions.append(tool)
+                    continue
+
+                existing_executor = existing.executor
+                replacement_executor = tool.executor
+                if (
+                    isinstance(existing_executor, MCPToolExecutor)
+                    and isinstance(replacement_executor, MCPToolExecutor)
+                    and existing_executor.client is replacement_executor.client
+                ):
+                    replacements.append(tool)
+                else:
+                    conflicts.add(tool.name)
+
+            if conflicts:
+                raise ValueError(
+                    "Dynamically advertised MCP tools conflict with existing runtime "
+                    f"tools: {sorted(conflicts)}"
+                )
 
-        self.add_runtime_tools(additions)
-        for tool in replacements:
-            self._tools[tool.name] = tool
+            self.add_runtime_tools(additions)
+            for tool in replacements:
+                self._tools[tool.name] = tool
 
         if additions:
             logger.info(
@@ -940,6 +946,46 @@ def _on_mcp_tools_changed(self, tools: Sequence[ToolDefinition]) -> None:
                 ", ".join(tool.name for tool in replacements),
             )
 
+    def _on_mcp_tools_reconciled(
+        self,
+        client: MCPClient,
+        tools: Sequence[MCPToolDefinition],
+    ) -> None:
+        """Replace this MCP client's tools with its current server snapshot."""
+        tool_names = [tool.name for tool in tools]
+        if len(tool_names) != len(set(tool_names)):
+            duplicates = {
+                name for name, count in Counter(tool_names).items() if count > 1
+            }
+            raise ValueError(f"Duplicate MCP tool names found: {duplicates}")
+
+        if self.filter_tools_regex:
+            pattern = re.compile(self.filter_tools_regex)
+            tools = [tool for tool in tools if pattern.match(tool.name)]
+            tool_names = [tool.name for tool in tools]
+
+        with self._tools_lock:
+            owned_names = {
+                name
+                for name, tool in self._tools.items()
+                if isinstance(tool.executor, MCPToolExecutor)
+                and tool.executor.client is client
+            }
+            conflicts = (set(tool_names) & set(self._tools)) - owned_names
+            if conflicts:
+                raise ValueError(
+                    "Dynamically advertised MCP tools conflict with existing runtime "
+                    f"tools: {sorted(conflicts)}"
+                )
+
+            reconciled = {
+                name: tool
+                for name, tool in self._tools.items()
+                if name not in owned_names
+            }
+            reconciled.update((tool.name, tool) for tool in tools)
+            object.__setattr__(self, "_tools", reconciled)
+
     @property
     def tools_map(self) -> dict[str, ToolDefinition]:
         """Get the initialized tools map.
@@ -949,7 +995,8 @@ def tools_map(self) -> dict[str, ToolDefinition]:
         if not self._initialized:
             raise RuntimeError("Agent not initialized; call _initialize() before use")
         # Isolate readers from background MCP tool updates.
-        return dict(self._tools)
+        with self._tools_lock:
+            return dict(self._tools)
 
     # -- Capability helpers -----------------------------------------------
     # Downstream code should branch on these properties rather than doing

openhands-sdk/openhands/sdk/conversation/impl/local_conversation.py

@@ -59,16 +59,19 @@
 from openhands.sdk.llm.llm_registry import LLMRegistry
 from openhands.sdk.logger import get_logger
 from openhands.sdk.marketplace.registry import MarketplaceRegistry
+from openhands.sdk.mcp.client import MCPClient
 from openhands.sdk.mcp.config import (
     MCPServer,
     coerce_mcp_config,
     dump_mcp_config,
     enabled_mcp_servers,
 )
+from openhands.sdk.mcp.tool import MCPToolDefinition
 from openhands.sdk.mcp.utils import (
     DefaultMCPToolProvider,
     MCPToolProvider,
     ToolsChangedCallback,
+    ToolsReconciledCallback,
 )
 from openhands.sdk.observability.laminar import observe
 from openhands.sdk.plugin import (
@@ -1277,6 +1280,7 @@ def _runtime_mcp_tools(
         mcp_config: dict[str, MCPServer],
         *,
         on_tools_changed: ToolsChangedCallback | None = None,
+        on_tools_reconciled: ToolsReconciledCallback | None = None,
     ) -> list[ToolDefinition]:
         # Servers the user switched off stay in the settings map but must not
         # be connected to. Filter before the emptiness check so an all-disabled
@@ -1289,14 +1293,23 @@ def _runtime_mcp_tools(
             _RUNTIME_MCP_TIMEOUT_SECS,
             on_tools_changed=on_tools_changed,
         )
+        client._tools_reconciled_callback = on_tools_reconciled
         return list(client.tools)
 
+    def _on_mcp_tools_reconciled(
+        self,
+        client: MCPClient,
+        tools: Sequence[MCPToolDefinition],
+    ) -> None:
+        self.agent._on_mcp_tools_reconciled(client, tools)
+
     def _runtime_mcp_tools_for_agent(self) -> list[ToolDefinition]:
         if not self.agent.supports_openhands_tools or not self.agent.mcp_config:
             return []
         return self._runtime_mcp_tools(
             self.agent.mcp_config,
-            on_tools_changed=self.agent._on_mcp_tools_changed,
+            on_tools_changed=lambda tools: self.agent._on_mcp_tools_changed(tools),
+            on_tools_reconciled=self._on_mcp_tools_reconciled,
         )
 
     def _runtime_skill_tools_for_agent(self) -> list[ToolDefinition]:
@@ -1366,7 +1379,13 @@ def load_plugin(self, plugin_ref: str) -> None:
             )
             merged_mcp = coerce_mcp_config(expanded_mcp["mcpServers"])
         runtime_mcp_tools = (
-            self._runtime_mcp_tools(runtime_plugin_mcp) if self._agent_ready else []
+            self._runtime_mcp_tools(
+                runtime_plugin_mcp,
+                on_tools_changed=lambda tools: self.agent._on_mcp_tools_changed(tools),
+                on_tools_reconciled=self._on_mcp_tools_reconciled,
+            )
+            if self._agent_ready
+            else []
         )
 
         with self._state:

openhands-sdk/openhands/sdk/mcp/client.py

@@ -2,7 +2,7 @@
 
 import asyncio
 import inspect
-from collections.abc import Callable, Iterator
+from collections.abc import Callable, Iterator, Sequence
 from typing import TYPE_CHECKING, Any
 
 from fastmcp import Client as AsyncMCPClient
@@ -15,6 +15,12 @@
     from openhands.sdk.mcp.tool import MCPToolDefinition
 
 
+ToolsReconciledCallback = Callable[
+    ["MCPClient", Sequence["MCPToolDefinition"]],
+    None,
+]
+
+
 class MCPClient(AsyncMCPClient):
     """MCP client with sync helpers and lifecycle management.
 
@@ -35,12 +41,14 @@ class MCPClient(AsyncMCPClient):
     _executor: AsyncExecutor
     _closed: bool
     _tools: "list[MCPToolDefinition]"
+    _tools_reconciled_callback: ToolsReconciledCallback | None
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._executor = AsyncExecutor()
         self._closed = False
         self._tools = []
+        self._tools_reconciled_callback = None
 
     @property
     def tools(self) -> "list[MCPToolDefinition]":

openhands-sdk/openhands/sdk/mcp/tool.py

@@ -1,6 +1,7 @@
 """Utility functions for MCP integration."""
 
 import copy
+import json
 import re
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
@@ -195,7 +196,7 @@ def close(self) -> None:
         self.client.sync_close()
 
 
-_mcp_dynamic_action_type: dict[str, type[Schema]] = {}
+_mcp_dynamic_action_type: dict[tuple[str, str], type[Schema]] = {}
 
 
 def _create_mcp_action_type(action_type: mcp.types.Tool) -> type[Schema]:
@@ -213,14 +214,17 @@ def _create_mcp_action_type(action_type: mcp.types.Tool) -> type[Schema]:
     to openai tool schema.
     """
 
-    # Tool.name should be unique, so we can cache the created types.
-    mcp_action_type = _mcp_dynamic_action_type.get(action_type.name)
+    cache_key = (
+        action_type.name,
+        json.dumps(action_type.inputSchema, sort_keys=True, separators=(",", ":")),
+    )
+    mcp_action_type = _mcp_dynamic_action_type.get(cache_key)
     if mcp_action_type:
         return mcp_action_type
 
     model_name = f"MCP{to_camel_case(action_type.name)}Action"
     mcp_action_type = Schema.from_mcp_schema(model_name, action_type.inputSchema)
-    _mcp_dynamic_action_type[action_type.name] = mcp_action_type
+    _mcp_dynamic_action_type[cache_key] = mcp_action_type
     return mcp_action_type