feat(mcp): adopt mcp__server__tool naming convention

Port from anomalyco/opencode#33533. Native MCP tools now register as
mcp__<server>__<tool> (double-underscore delimiter) instead of
mcp_<server>_<tool>, aligning with the convention used by Claude Code,
Codex, and OpenCode.

The double-underscore delimiter disambiguates the server/tool boundary
even when either component contains underscores (the single-underscore
form was ambiguous, which is why is_mcp_tool_parallel_safe already had to
track provenance in a side-map). It also unifies native registration with
the Anthropic-OAuth wire form (_MCP_TOOL_PREFIX = 'mcp__'), so the
single->double promotion that path performed is now a no-op for native
tools while still handling legacy replayed names.

- tools/mcp_tool.py: add MCP_TOOL_NAME_PREFIX + mcp_prefixed_tool_name()
  helper; route _convert_mcp_schema, utility schemas, refresh stale-set,
  and the parallel-safe prefix gate through it
- agent/transports/codex_event_projector.py: mirror convention in the
  deterministic call_id input for MCP server-executed tool calls
- tests: update produced-name assertions to the new convention

agent/transports/codex_event_projector.py

@@ -217,7 +217,9 @@ class CodexEventProjector:
     def _project_mcp_tool_call(self, item: dict, item_id: str) -> ProjectionResult:
         server = item.get("server") or "mcp"
         tool = item.get("tool") or "unknown"
-        call_id = _deterministic_call_id(f"mcp_{server}_{tool}", item_id)
+        # Mirror the native MCP tool-name convention (mcp__server__tool) so the
+        # deterministic call_id input stays consistent with registration names.
+        call_id = _deterministic_call_id(f"mcp__{server}__{tool}", item_id)
         args = item.get("arguments") or {}
         if not isinstance(args, dict):
             args = {"arguments": args}

tests/test_toolsets.py

@@ -133,9 +133,9 @@ class TestValidateToolset:
     def test_mcp_alias_uses_live_registry(self, monkeypatch):
         reg = ToolRegistry()
         reg.register(
-            name="mcp_dynserver_ping",
+            name="mcp__dynserver__ping",
             toolset="mcp-dynserver",
-            schema=_make_schema("mcp_dynserver_ping", "Ping"),
+            schema=_make_schema("mcp__dynserver__ping", "Ping"),
             handler=_dummy_handler,
         )
         reg.register_toolset_alias("dynserver", "mcp-dynserver")
@@ -144,7 +144,7 @@ class TestValidateToolset:
 
         assert validate_toolset("dynserver") is True
         assert validate_toolset("mcp-dynserver") is True
-        assert "mcp_dynserver_ping" in resolve_toolset("dynserver")
+        assert "mcp__dynserver__ping" in resolve_toolset("dynserver")
 
 
 class TestGetToolsetInfo:

tests/tools/test_mcp_dynamic_discovery.py

@@ -30,10 +30,10 @@ class TestRegisterServerTools:
 
         with patch("tools.registry.registry", mock_registry):
             registered = _register_server_tools("my_srv", server, {})
-            assert "mcp_my_srv_my_tool" in registered
-            assert "mcp_my_srv_my_tool" in mock_registry.get_all_tool_names()
+            assert "mcp__my_srv__my_tool" in registered
+            assert "mcp__my_srv__my_tool" in mock_registry.get_all_tool_names()
             assert validate_toolset("my_srv") is True
-            assert "mcp_my_srv_my_tool" in resolve_toolset("my_srv")
+            assert "mcp__my_srv__my_tool" in resolve_toolset("my_srv")
 
 
 class TestRefreshTools:
@@ -53,11 +53,11 @@ class TestRefreshTools:
 
         # Seed initial state: one old tool registered
         mock_registry.register(
-            name="mcp_live_srv_old_tool", toolset="mcp-live_srv", schema={},
+            name="mcp__live_srv__old_tool", toolset="mcp-live_srv", schema={},
             handler=lambda x: x, check_fn=lambda: True, is_async=False,
             description="", emoji="",
         )
-        server._registered_tool_names = ["mcp_live_srv_old_tool"]
+        server._registered_tool_names = ["mcp__live_srv__old_tool"]
 
         # New tool list from server
         new_tool = _make_mcp_tool("new_tool", "new behavior")
@@ -69,11 +69,11 @@ class TestRefreshTools:
 
         with patch("tools.registry.registry", mock_registry):
             await server._refresh_tools()
-            assert "mcp_live_srv_old_tool" not in mock_registry.get_all_tool_names()
-            assert "mcp_live_srv_old_tool" not in resolve_toolset("live_srv")
-            assert "mcp_live_srv_new_tool" in mock_registry.get_all_tool_names()
-            assert "mcp_live_srv_new_tool" in resolve_toolset("live_srv")
-            assert server._registered_tool_names == ["mcp_live_srv_new_tool"]
+            assert "mcp__live_srv__old_tool" not in mock_registry.get_all_tool_names()
+            assert "mcp__live_srv__old_tool" not in resolve_toolset("live_srv")
+            assert "mcp__live_srv__new_tool" in mock_registry.get_all_tool_names()
+            assert "mcp__live_srv__new_tool" in resolve_toolset("live_srv")
+            assert server._registered_tool_names == ["mcp__live_srv__new_tool"]
 
 
 class TestMessageHandler:

tests/tools/test_mcp_tool.py

@@ -143,7 +143,7 @@ class TestSchemaConversion:
         mcp_tool = _make_mcp_tool(name="read_file", description="Read a file")
         schema = _convert_mcp_schema("filesystem", mcp_tool)
 
-        assert schema["name"] == "mcp_filesystem_read_file"
+        assert schema["name"] == "mcp__filesystem__read_file"
         assert schema["description"] == "Read a file"
         assert "properties" in schema["parameters"]
 
@@ -376,7 +376,7 @@ class TestSchemaConversion:
         bare_tool = types.SimpleNamespace(name="probe", description="Probe")
         schema = _convert_mcp_schema("srv", bare_tool)
 
-        assert schema["name"] == "mcp_srv_probe"
+        assert schema["name"] == "mcp__srv__probe"
         assert schema["parameters"] == {"type": "object", "properties": {}}
 
     def test_convert_mcp_schema_with_none_inputschema(self):
@@ -398,7 +398,7 @@ class TestSchemaConversion:
         mcp_tool = _make_mcp_tool(name="list_dir")
         schema = _convert_mcp_schema("my_server", mcp_tool)
 
-        assert schema["name"] == "mcp_my_server_list_dir"
+        assert schema["name"] == "mcp__my_server__list_dir"
 
     def test_hyphens_sanitized_to_underscores(self):
         """Hyphens in tool/server names are replaced with underscores for LLM compat."""
@@ -407,7 +407,7 @@ class TestSchemaConversion:
         mcp_tool = _make_mcp_tool(name="get-sum")
         schema = _convert_mcp_schema("my-server", mcp_tool)
 
-        assert schema["name"] == "mcp_my_server_get_sum"
+        assert schema["name"] == "mcp__my_server__get_sum"
         assert "-" not in schema["name"]
 
 
@@ -736,10 +736,10 @@ class TestDiscoverAndRegister:
                 _discover_and_register_server("fs", {"command": "npx", "args": []})
             )
 
-        assert "mcp_fs_read_file" in registered
-        assert "mcp_fs_write_file" in registered
-        assert "mcp_fs_read_file" in mock_registry.get_all_tool_names()
-        assert "mcp_fs_write_file" in mock_registry.get_all_tool_names()
+        assert "mcp__fs__read_file" in registered
+        assert "mcp__fs__write_file" in registered
+        assert "mcp__fs__read_file" in mock_registry.get_all_tool_names()
+        assert "mcp__fs__write_file" in mock_registry.get_all_tool_names()
 
         _servers.pop("fs", None)
 
@@ -767,8 +767,8 @@ class TestDiscoverAndRegister:
 
             assert validate_toolset("myserver") is True
             assert validate_toolset("mcp-myserver") is True
-            assert "mcp_myserver_ping" in resolve_toolset("myserver")
-            assert "mcp_myserver_ping" in resolve_toolset("mcp-myserver")
+            assert "mcp__myserver__ping" in resolve_toolset("myserver")
+            assert "mcp__myserver__ping" in resolve_toolset("mcp-myserver")
 
         _servers.pop("myserver", None)
 
@@ -793,9 +793,9 @@ class TestDiscoverAndRegister:
                 _discover_and_register_server("srv", {"command": "test"})
             )
 
-        entry = mock_registry._tools.get("mcp_srv_do_thing")
+        entry = mock_registry._tools.get("mcp__srv__do_thing")
         assert entry is not None
-        assert entry.schema["name"] == "mcp_srv_do_thing"
+        assert entry.schema["name"] == "mcp__srv__do_thing"
         assert "parameters" in entry.schema
         assert entry.is_async is False
         assert entry.toolset == "mcp-srv"
@@ -876,7 +876,7 @@ class TestMCPServerTask:
         server = MCPServerTask("srv")
         server._config = {"command": "test"}
         server._tools = [_make_mcp_tool("old"), _make_mcp_tool("keep")]
-        server._registered_tool_names = ["mcp_srv_old", "mcp_srv_keep"]
+        server._registered_tool_names = ["mcp__srv__old", "mcp__srv__keep"]
         server.session = MagicMock()
         server.session.list_tools = AsyncMock(
             return_value=SimpleNamespace(tools=[_make_mcp_tool("keep"), _make_mcp_tool("new")])
@@ -884,31 +884,31 @@ class TestMCPServerTask:
 
         with patch("tools.registry.registry", mock_registry):
             mock_registry.register(
-                name="mcp_srv_old",
+                name="mcp__srv__old",
                 toolset="mcp-srv",
-                schema={"name": "mcp_srv_old", "description": "Old"},
+                schema={"name": "mcp__srv__old", "description": "Old"},
                 handler=lambda *_args, **_kwargs: "{}",
             )
             mock_registry.register(
-                name="mcp_srv_keep",
+                name="mcp__srv__keep",
                 toolset="mcp-srv",
-                schema={"name": "mcp_srv_keep", "description": "Keep"},
+                schema={"name": "mcp__srv__keep", "description": "Keep"},
                 handler=lambda *_args, **_kwargs: "{}",
             )
 
             asyncio.run(server._refresh_tools())
 
             names = mock_registry.get_all_tool_names()
-            assert "mcp_srv_old" not in names
-            assert "mcp_srv_keep" in names
-            assert "mcp_srv_new" in names
+            assert "mcp__srv__old" not in names
+            assert "mcp__srv__keep" in names
+            assert "mcp__srv__new" in names
             assert set(server._registered_tool_names) == {
-                "mcp_srv_keep",
-                "mcp_srv_new",
-                "mcp_srv_list_resources",
-                "mcp_srv_read_resource",
-                "mcp_srv_list_prompts",
-                "mcp_srv_get_prompt",
+                "mcp__srv__keep",
+                "mcp__srv__new",
+                "mcp__srv__list_resources",
+                "mcp__srv__read_resource",
+                "mcp__srv__list_prompts",
+                "mcp__srv__get_prompt",
             }
 
     def test_schedule_tools_refresh_keeps_task_until_done(self):
@@ -1059,11 +1059,11 @@ class TestToolsetInjection:
             from tools.mcp_tool import discover_mcp_tools
             result = discover_mcp_tools()
 
-            assert "mcp_fs_list_files" in result
+            assert "mcp__fs__list_files" in result
             assert validate_toolset("fs") is True
             assert validate_toolset("mcp-fs") is True
-            assert "mcp_fs_list_files" in resolve_toolset("fs")
-            assert "mcp_fs_list_files" in resolve_toolset("mcp-fs")
+            assert "mcp__fs__list_files" in resolve_toolset("fs")
+            assert "mcp__fs__list_files" in resolve_toolset("mcp-fs")
 
     def test_server_toolset_skips_builtin_collision(self):
         """MCP raw aliases never overwrite a built-in toolset name."""
@@ -1099,9 +1099,9 @@ class TestToolsetInjection:
             discover_mcp_tools()
 
             assert fake_toolsets["terminal"]["description"] == "Terminal tools"
-            assert "mcp_terminal_run" not in resolve_toolset("terminal")
+            assert "mcp__terminal__run" not in resolve_toolset("terminal")
             assert validate_toolset("mcp-terminal") is True
-            assert "mcp_terminal_run" in resolve_toolset("mcp-terminal")
+            assert "mcp__terminal__run" in resolve_toolset("mcp-terminal")
 
     def test_server_connection_failure_skipped(self):
         """If one server fails to connect, others still proceed."""
@@ -1139,8 +1139,8 @@ class TestToolsetInjection:
             from tools.mcp_tool import discover_mcp_tools
             result = discover_mcp_tools()
 
-        assert "mcp_good_ping" in result
-        assert "mcp_broken_ping" not in result
+        assert "mcp__good__ping" in result
+        assert "mcp__broken__ping" not in result
         assert call_count == 2
 
     def test_partial_failure_retry_on_second_call(self):
@@ -1182,8 +1182,8 @@ class TestToolsetInjection:
 
             # First call: good connects, broken fails
             result1 = discover_mcp_tools()
-            assert "mcp_good_ping" in result1
-            assert "mcp_broken_ping" not in result1
+            assert "mcp__good__ping" in result1
+            assert "mcp__broken__ping" not in result1
             first_attempts = call_count
 
             # "Fix" the broken server
@@ -1192,8 +1192,8 @@ class TestToolsetInjection:
 
             # Second call: should retry broken, skip good
             result2 = discover_mcp_tools()
-            assert "mcp_good_ping" in result2
-            assert "mcp_broken_ping" in result2
+            assert "mcp__good__ping" in result2
+            assert "mcp__broken__ping" in result2
             assert call_count == 1  # Only broken retried
 
 
@@ -1261,10 +1261,10 @@ class TestShutdown:
 
         _servers.clear()
         registry.register(
-            name="mcp_test_ping",
+            name="mcp__test__ping",
             toolset="mcp-test",
             schema={
-                "name": "mcp_test_ping",
+                "name": "mcp__test__ping",
                 "description": "Ping",
                 "parameters": {"type": "object", "properties": {}},
             },
@@ -1273,19 +1273,19 @@ class TestShutdown:
         registry.register_toolset_alias("test", "mcp-test")
 
         server = MCPServerTask("test")
-        server._registered_tool_names = ["mcp_test_ping"]
+        server._registered_tool_names = ["mcp__test__ping"]
         _servers["test"] = server
 
         mcp_mod._ensure_mcp_loop()
         try:
             assert validate_toolset("test") is True
-            assert "mcp_test_ping" in resolve_toolset("test")
+            assert "mcp__test__ping" in resolve_toolset("test")
             shutdown_mcp_servers()
         finally:
             mcp_mod._mcp_loop = None
             mcp_mod._mcp_thread = None
 
-        assert "mcp_test_ping" not in registry.get_all_tool_names()
+        assert "mcp__test__ping" not in registry.get_all_tool_names()
         assert validate_toolset("test") is False
 
     def test_shutdown_handles_errors(self):
@@ -1961,10 +1961,10 @@ class TestUtilitySchemas:
         schemas = _build_utility_schemas("myserver")
         assert len(schemas) == 4
         names = [s["schema"]["name"] for s in schemas]
-        assert "mcp_myserver_list_resources" in names
-        assert "mcp_myserver_read_resource" in names
-        assert "mcp_myserver_list_prompts" in names
-        assert "mcp_myserver_get_prompt" in names
+        assert "mcp__myserver__list_resources" in names
+        assert "mcp__myserver__read_resource" in names
+        assert "mcp__myserver__list_prompts" in names
+        assert "mcp__myserver__get_prompt" in names
 
     def test_hyphens_sanitized_in_utility_names(self):
         from tools.mcp_tool import _build_utility_schemas
@@ -1973,7 +1973,7 @@ class TestUtilitySchemas:
         names = [s["schema"]["name"] for s in schemas]
         for name in names:
             assert "-" not in name
-        assert "mcp_my_server_list_resources" in names
+        assert "mcp__my_server__list_resources" in names
 
     def test_list_resources_schema_no_required_params(self):
         from tools.mcp_tool import _build_utility_schemas
@@ -2296,11 +2296,11 @@ class TestUtilityToolRegistration:
             )
 
         # Regular tool + 4 utility tools
-        assert "mcp_fs_read_file" in registered
-        assert "mcp_fs_list_resources" in registered
-        assert "mcp_fs_read_resource" in registered
-        assert "mcp_fs_list_prompts" in registered
-        assert "mcp_fs_get_prompt" in registered
+        assert "mcp__fs__read_file" in registered
+        assert "mcp__fs__list_resources" in registered
+        assert "mcp__fs__read_resource" in registered
+        assert "mcp__fs__list_prompts" in registered
+        assert "mcp__fs__get_prompt" in registered
         assert len(registered) == 5
 
         # All in the registry
@@ -2331,8 +2331,8 @@ class TestUtilityToolRegistration:
             )
 
         # Check that utility tools are in the right toolset
-        for tool_name in ["mcp_myserv_list_resources", "mcp_myserv_read_resource",
-                          "mcp_myserv_list_prompts", "mcp_myserv_get_prompt"]:
+        for tool_name in ["mcp__myserv__list_resources", "mcp__myserv__read_resource",
+                          "mcp__myserv__list_prompts", "mcp__myserv__get_prompt"]:
             entry = mock_registry._tools.get(tool_name)
             assert entry is not None, f"{tool_name} not found in registry"
             assert entry.toolset == "mcp-myserv"
@@ -2359,7 +2359,7 @@ class TestUtilityToolRegistration:
                 _discover_and_register_server("chk", {"command": "test"})
             )
 
-        entry = mock_registry._tools.get("mcp_chk_list_resources")
+        entry = mock_registry._tools.get("mcp__chk__list_resources")
         assert entry is not None
         # Server is connected, check_fn should return True
         assert entry.check_fn() is True
@@ -3284,12 +3284,12 @@ class TestDiscoveryFailedCount:
             server.session = MagicMock()
             server._tools = [_make_mcp_tool("tool_a")]
             _servers[name] = server
-            return [f"mcp_{name}_tool_a"]
+            return [f"mcp__{name}__tool_a"]
 
         with patch("tools.mcp_tool._load_mcp_config", return_value=fake_config), \
              patch("tools.mcp_tool._discover_and_register_server", side_effect=fake_register), \
              patch("tools.mcp_tool._MCP_AVAILABLE", True), \
-             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp_good_server_tool_a"]):
+             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp__good_server__tool_a"]):
             _ensure_mcp_loop()
 
             # Capture the logger to verify failed_count in summary
@@ -3358,12 +3358,12 @@ class TestDiscoveryFailedCount:
             server.session = MagicMock()
             server._tools = [_make_mcp_tool("t")]
             _servers[name] = server
-            return [f"mcp_{name}_t"]
+            return [f"mcp__{name}__t"]
 
         with patch("tools.mcp_tool._load_mcp_config", return_value=fake_config), \
              patch("tools.mcp_tool._discover_and_register_server", side_effect=selective_register), \
              patch("tools.mcp_tool._MCP_AVAILABLE", True), \
-             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp_ok1_t", "mcp_ok2_t"]):
+             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp__ok1__t", "mcp__ok2__t"]):
             _ensure_mcp_loop()
 
             with patch("tools.mcp_tool.logger") as mock_logger:
@@ -3430,7 +3430,7 @@ class TestMCPSelectiveToolLoading:
             config,
             session=SimpleNamespace(),
         )
-        assert registered == ["mcp_ink_create_service"]
+        assert registered == ["mcp__ink__create_service"]
 
     def test_exclude_filter_registers_all_except_listed_tools(self):
         config = {
@@ -3444,8 +3444,8 @@ class TestMCPSelectiveToolLoading:
             session=SimpleNamespace(),
         )
         assert registered == [
-            "mcp_ink_exclude_create_service",
-            "mcp_ink_exclude_list_services",
+            "mcp__ink_exclude__create_service",
+            "mcp__ink_exclude__list_services",
         ]
 
     def test_include_filter_skips_utility_tools_without_capabilities(self):
@@ -3459,8 +3459,8 @@ class TestMCPSelectiveToolLoading:
             config,
             session=SimpleNamespace(),
         )
-        assert registered == ["mcp_ink_no_caps_create_service"]
-        assert set(mock_registry.get_all_tool_names()) == {"mcp_ink_no_caps_create_service"}
+        assert registered == ["mcp__ink_no_caps__create_service"]
+        assert set(mock_registry.get_all_tool_names()) == {"mcp__ink_no_caps__create_service"}
 
     def test_no_filter_registers_all_server_tools_when_no_utilities_supported(self):
         registered, _ = self._run_discover(
@@ -3470,9 +3470,9 @@ class TestMCPSelectiveToolLoading:
             session=SimpleNamespace(),
         )
         assert registered == [
-            "mcp_ink_no_filter_create_service",
-            "mcp_ink_no_filter_delete_service",
-            "mcp_ink_no_filter_list_services",
+            "mcp__ink_no_filter__create_service",
+            "mcp__ink_no_filter__delete_service",
+            "mcp__ink_no_filter__list_services",
         ]
 
     def test_resources_and_prompts_can_be_disabled_explicitly(self):
@@ -3495,7 +3495,7 @@ class TestMCPSelectiveToolLoading:
             config,
             session=session,
         )
-        assert registered == ["mcp_ink_disabled_utils_create_service"]
+        assert registered == ["mcp__ink_disabled_utils__create_service"]
 
     def test_registers_only_utility_tools_supported_by_server_capabilities(self):
         session = SimpleNamespace(
@@ -3508,11 +3508,11 @@ class TestMCPSelectiveToolLoading:
             {"url": "https://mcp.example.com"},
             session=session,
         )
-        assert "mcp_ink_resources_only_create_service" in registered
-        assert "mcp_ink_resources_only_list_resources" in registered
-        assert "mcp_ink_resources_only_read_resource" in registered
-        assert "mcp_ink_resources_only_list_prompts" not in registered
-        assert "mcp_ink_resources_only_get_prompt" not in registered
+        assert "mcp__ink_resources_only__create_service" in registered
+        assert "mcp__ink_resources_only__list_resources" in registered
+        assert "mcp__ink_resources_only__read_resource" in registered
+        assert "mcp__ink_resources_only__list_prompts" not in registered
+        assert "mcp__ink_resources_only__get_prompt" not in registered
 
     def test_existing_tool_names_reflect_registered_subset(self):
         from tools.mcp_tool import _existing_tool_names, _servers, _discover_and_register_server
@@ -3541,8 +3541,8 @@ class TestMCPSelectiveToolLoading:
 
         try:
             registered, existing = asyncio.run(run())
-            assert registered == ["mcp_ink_existing_create_service"]
-            assert existing == ["mcp_ink_existing_create_service"]
+            assert registered == ["mcp__ink_existing__create_service"]
+            assert existing == ["mcp__ink_existing__create_service"]
         finally:
             _servers.pop("ink_existing", None)
 
@@ -3667,12 +3667,12 @@ class TestMCPBuiltinCollisionGuard:
         # Pre-register a "built-in" tool with the name that the MCP tool would produce.
         # Server "abc", tool "search" → mcp_abc_search
         builtin_schema = {
-            "name": "mcp_abc_search",
+            "name": "mcp__abc__search",
             "description": "A hypothetical built-in",
             "parameters": {"type": "object", "properties": {}},
         }
         mock_registry.register(
-            name="mcp_abc_search", toolset="web",
+            name="mcp__abc__search", toolset="web",
             schema=builtin_schema, handler=lambda a, **k: "{}",
         )
 
@@ -3692,8 +3692,8 @@ class TestMCPBuiltinCollisionGuard:
             )
 
         # The MCP tool should have been skipped — built-in preserved.
-        assert "mcp_abc_search" not in registered
-        assert mock_registry.get_toolset_for_tool("mcp_abc_search") == "web"
+        assert "mcp__abc__search" not in registered
+        assert mock_registry.get_toolset_for_tool("mcp__abc__search") == "web"
 
         _servers.pop("abc", None)
 
@@ -3718,8 +3718,8 @@ class TestMCPBuiltinCollisionGuard:
                 _discover_and_register_server("minimax", {"command": "test", "args": []})
             )
 
-        assert "mcp_minimax_web_search" in registered
-        assert mock_registry.get_toolset_for_tool("mcp_minimax_web_search") == "mcp-minimax"
+        assert "mcp__minimax__web_search" in registered
+        assert mock_registry.get_toolset_for_tool("mcp__minimax__web_search") == "mcp-minimax"
 
         _servers.pop("minimax", None)
 
@@ -3732,12 +3732,12 @@ class TestMCPBuiltinCollisionGuard:
 
         # Pre-register an MCP tool from a different server.
         mcp_schema = {
-            "name": "mcp_srv_do_thing",
+            "name": "mcp__srv__do_thing",
             "description": "From another MCP server",
             "parameters": {"type": "object", "properties": {}},
         }
         mock_registry.register(
-            name="mcp_srv_do_thing", toolset="mcp-old",
+            name="mcp__srv__do_thing", toolset="mcp-old",
             schema=mcp_schema, handler=lambda a, **k: "{}",
         )
 
@@ -3757,8 +3757,8 @@ class TestMCPBuiltinCollisionGuard:
             )
 
         # MCP-to-MCP collision is allowed — the new server wins.
-        assert "mcp_srv_do_thing" in registered
-        assert mock_registry.get_toolset_for_tool("mcp_srv_do_thing") == "mcp-srv"
+        assert "mcp__srv__do_thing" in registered
+        assert mock_registry.get_toolset_for_tool("mcp__srv__do_thing") == "mcp-srv"
 
         _servers.pop("srv", None)
 
@@ -3805,7 +3805,7 @@ class TestSanitizeMcpNameComponent:
 
         mcp_tool = _make_mcp_tool(name="search")
         schema = _convert_mcp_schema("ai.exa/exa", mcp_tool)
-        assert schema["name"] == "mcp_ai_exa_exa_search"
+        assert schema["name"] == "mcp__ai_exa_exa__search"
         # Must match Anthropic's pattern: ^[a-zA-Z0-9_-]{1,128}$
         import re
         assert re.match(r"^[a-zA-Z0-9_-]{1,128}$", schema["name"])
@@ -3827,16 +3827,16 @@ class TestSanitizeMcpNameComponent:
 
         reg = ToolRegistry()
         reg.register(
-            name="mcp_ai_exa_exa_search",
+            name="mcp__ai_exa_exa__search",
             toolset="mcp-ai.exa/exa",
-            schema={"name": "mcp_ai_exa_exa_search", "description": "Search", "parameters": {"type": "object", "properties": {}}},
+            schema={"name": "mcp__ai_exa_exa__search", "description": "Search", "parameters": {"type": "object", "properties": {}}},
             handler=lambda *_args, **_kwargs: "{}",
         )
         reg.register_toolset_alias("ai.exa/exa", "mcp-ai.exa/exa")
 
         with patch("tools.registry.registry", reg):
             assert validate_toolset("ai.exa/exa") is True
-            assert "mcp_ai_exa_exa_search" in resolve_toolset("ai.exa/exa")
+            assert "mcp__ai_exa_exa__search" in resolve_toolset("ai.exa/exa")
 
 
 # ---------------------------------------------------------------------------
@@ -3869,9 +3869,9 @@ class TestRegisterMcpServers:
 
         try:
             with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
-                 patch("tools.mcp_tool._existing_tool_names", return_value=["mcp_existing_tool"]):
+                 patch("tools.mcp_tool._existing_tool_names", return_value=["mcp__existing__tool"]):
                 result = register_mcp_servers({"existing": {"command": "test"}})
-            assert result == ["mcp_existing_tool"]
+            assert result == ["mcp__existing__tool"]
         finally:
             _servers.pop("existing", None)
 
@@ -3893,17 +3893,17 @@ class TestRegisterMcpServers:
 
         async def fake_register(name, cfg):
             server = _make_mock_server(name)
-            server._registered_tool_names = ["mcp_my_server_tool1"]
+            server._registered_tool_names = ["mcp__my_server__tool1"]
             _servers[name] = server
-            return ["mcp_my_server_tool1"]
+            return ["mcp__my_server__tool1"]
 
         with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
              patch("tools.mcp_tool._discover_and_register_server", side_effect=fake_register), \
-             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp_my_server_tool1"]):
+             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp__my_server__tool1"]):
             _ensure_mcp_loop()
             result = register_mcp_servers(fake_config)
 
-        assert "mcp_my_server_tool1" in result
+        assert "mcp__my_server__tool1" in result
         _servers.pop("my_server", None)
 
     def test_logs_summary_on_success(self):
@@ -3913,13 +3913,13 @@ class TestRegisterMcpServers:
 
         async def fake_register(name, cfg):
             server = _make_mock_server(name)
-            server._registered_tool_names = ["mcp_srv_t1", "mcp_srv_t2"]
+            server._registered_tool_names = ["mcp__srv__t1", "mcp__srv__t2"]
             _servers[name] = server
-            return ["mcp_srv_t1", "mcp_srv_t2"]
+            return ["mcp__srv__t1", "mcp__srv__t2"]
 
         with patch("tools.mcp_tool._MCP_AVAILABLE", True), \
              patch("tools.mcp_tool._discover_and_register_server", side_effect=fake_register), \
-             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp_srv_t1", "mcp_srv_t2"]):
+             patch("tools.mcp_tool._existing_tool_names", return_value=["mcp__srv__t1", "mcp__srv__t2"]):
             _ensure_mcp_loop()
 
             with patch("tools.mcp_tool.logger") as mock_logger:
@@ -3957,7 +3957,7 @@ class TestMcpParallelToolCalls:
         with _lock:
             _parallel_safe_servers.clear()
             _mcp_tool_server_names.clear()
-        assert is_mcp_tool_parallel_safe("mcp_docs_search") is False
+        assert is_mcp_tool_parallel_safe("mcp__docs__search") is False
 
     def test_is_mcp_tool_parallel_safe_with_flag(self):
         """MCP tool from a parallel-safe server returns True."""
@@ -3967,20 +3967,20 @@ class TestMcpParallelToolCalls:
         )
         with _lock:
             _parallel_safe_servers.add("docs")
-            _mcp_tool_server_names["mcp_docs_search"] = "docs"
-            _mcp_tool_server_names["mcp_docs_read_file"] = "docs"
-            _mcp_tool_server_names["mcp_github_list_repos"] = "github"
+            _mcp_tool_server_names["mcp__docs__search"] = "docs"
+            _mcp_tool_server_names["mcp__docs__read_file"] = "docs"
+            _mcp_tool_server_names["mcp__github__list_repos"] = "github"
         try:
-            assert is_mcp_tool_parallel_safe("mcp_docs_search") is True
-            assert is_mcp_tool_parallel_safe("mcp_docs_read_file") is True
+            assert is_mcp_tool_parallel_safe("mcp__docs__search") is True
+            assert is_mcp_tool_parallel_safe("mcp__docs__read_file") is True
             # Different server should be False
-            assert is_mcp_tool_parallel_safe("mcp_github_list_repos") is False
+            assert is_mcp_tool_parallel_safe("mcp__github__list_repos") is False
         finally:
             with _lock:
                 _parallel_safe_servers.discard("docs")
-                _mcp_tool_server_names.pop("mcp_docs_search", None)
-                _mcp_tool_server_names.pop("mcp_docs_read_file", None)
-                _mcp_tool_server_names.pop("mcp_github_list_repos", None)
+                _mcp_tool_server_names.pop("mcp__docs__search", None)
+                _mcp_tool_server_names.pop("mcp__docs__read_file", None)
+                _mcp_tool_server_names.pop("mcp__github__list_repos", None)
 
     def test_is_mcp_tool_parallel_safe_server_with_underscores(self):
         """Server names containing underscores are correctly matched."""
@@ -3990,13 +3990,13 @@ class TestMcpParallelToolCalls:
         )
         with _lock:
             _parallel_safe_servers.add("my_server")
-            _mcp_tool_server_names["mcp_my_server_query"] = "my_server"
+            _mcp_tool_server_names["mcp__my_server__query"] = "my_server"
         try:
-            assert is_mcp_tool_parallel_safe("mcp_my_server_query") is True
+            assert is_mcp_tool_parallel_safe("mcp__my_server__query") is True
         finally:
             with _lock:
                 _parallel_safe_servers.discard("my_server")
-                _mcp_tool_server_names.pop("mcp_my_server_query", None)
+                _mcp_tool_server_names.pop("mcp__my_server__query", None)
 
     def test_is_mcp_tool_parallel_safe_uses_exact_registered_server(self):
         """Ambiguous MCP names must not match a shorter parallel-safe prefix."""
@@ -4006,16 +4006,16 @@ class TestMcpParallelToolCalls:
         )
         with _lock:
             _parallel_safe_servers.add("a")
-            _mcp_tool_server_names["mcp_a_search"] = "a"
-            _mcp_tool_server_names["mcp_a_b_tool"] = "a_b"
+            _mcp_tool_server_names["mcp__a__search"] = "a"
+            _mcp_tool_server_names["mcp__a_b__tool"] = "a_b"
         try:
-            assert is_mcp_tool_parallel_safe("mcp_a_search") is True
-            assert is_mcp_tool_parallel_safe("mcp_a_b_tool") is False
+            assert is_mcp_tool_parallel_safe("mcp__a__search") is True
+            assert is_mcp_tool_parallel_safe("mcp__a_b__tool") is False
         finally:
             with _lock:
                 _parallel_safe_servers.discard("a")
-                _mcp_tool_server_names.pop("mcp_a_search", None)
-                _mcp_tool_server_names.pop("mcp_a_b_tool", None)
+                _mcp_tool_server_names.pop("mcp__a__search", None)
+                _mcp_tool_server_names.pop("mcp__a_b__tool", None)
 
     def test_registered_tool_provenance_prevents_prefix_collision(self):
         """Registration records exact server ownership for ambiguous names."""
@@ -4031,22 +4031,22 @@ class TestMcpParallelToolCalls:
         )
         registered = _register_server_tools("a_b", server, {})
         try:
-            assert registered == ["mcp_a_b_tool"]
+            assert registered == ["mcp__a_b__tool"]
             with _lock:
-                assert _mcp_tool_server_names["mcp_a_b_tool"] == "a_b"
+                assert _mcp_tool_server_names["mcp__a_b__tool"] == "a_b"
                 _parallel_safe_servers.add("a")
-            assert is_mcp_tool_parallel_safe("mcp_a_b_tool") is False
+            assert is_mcp_tool_parallel_safe("mcp__a_b__tool") is False
 
             with _lock:
                 _parallel_safe_servers.add("a_b")
-            assert is_mcp_tool_parallel_safe("mcp_a_b_tool") is True
+            assert is_mcp_tool_parallel_safe("mcp__a_b__tool") is True
         finally:
             for tool_name in registered:
                 registry.deregister(tool_name)
             with _lock:
                 _parallel_safe_servers.discard("a")
                 _parallel_safe_servers.discard("a_b")
-                _mcp_tool_server_names.pop("mcp_a_b_tool", None)
+                _mcp_tool_server_names.pop("mcp__a_b__tool", None)
 
     def test_is_mcp_tool_parallel_safe_no_tool_suffix(self):
         """Tool name that is just 'mcp_{server}' without a tool part returns False."""
@@ -4057,12 +4057,12 @@ class TestMcpParallelToolCalls:
         with _lock:
             _parallel_safe_servers.add("docs")
             _mcp_tool_server_names.pop("mcp_docs", None)
-            _mcp_tool_server_names.pop("mcp_docs_", None)
+            _mcp_tool_server_names.pop("mcp__docs__", None)
         try:
             # "mcp_docs" has no tool part after the server name
             assert is_mcp_tool_parallel_safe("mcp_docs") is False
             # "mcp_docs_" has empty tool part
-            assert is_mcp_tool_parallel_safe("mcp_docs_") is False
+            assert is_mcp_tool_parallel_safe("mcp__docs__") is False
         finally:
             with _lock:
                 _parallel_safe_servers.discard("docs")

tools/mcp_tool.py

@@ -1602,8 +1602,7 @@ class MCPServerTask:
             # notifications. Tools absent from the fresh list are no longer
             # callable, so remove only those stale registry entries first.
             stale_tool_names = old_tool_names - {
-                f"mcp_{sanitize_mcp_name_component(self.name)}_"
-                f"{sanitize_mcp_name_component(tool.name)}"
+                mcp_prefixed_tool_name(self.name, tool.name)
                 for tool in new_mcp_tools
             }
             for tool_name in stale_tool_names:
@@ -3678,6 +3677,27 @@ def sanitize_mcp_name_component(value: str) -> str:
     return re.sub(r"[^A-Za-z0-9_]", "_", str(value or ""))
 
 
+# Native MCP tool-name prefix. Hermes uses the ``mcp__<server>__<tool>``
+# convention shared by Claude Code, Codex, and OpenCode (anomalyco/opencode
+# #33533). The double-underscore delimiter disambiguates the server/tool
+# boundary even when either component contains underscores, and matches the
+# naming models are trained on. It also aligns native registration with the
+# Anthropic-OAuth wire form (``_MCP_TOOL_PREFIX`` in anthropic_adapter.py),
+# removing the single->double rewrite that path previously had to perform.
+MCP_TOOL_NAME_PREFIX = "mcp__"
+_MCP_NAME_DELIM = "__"
+
+
+def mcp_prefixed_tool_name(server_name: str, tool_name: str) -> str:
+    """Build the registry/wire name for an MCP tool.
+
+    Produces ``mcp__<sanitizedServer>__<sanitizedTool>``.
+    """
+    safe_server = sanitize_mcp_name_component(server_name)
+    safe_tool = sanitize_mcp_name_component(tool_name)
+    return f"{MCP_TOOL_NAME_PREFIX}{safe_server}{_MCP_NAME_DELIM}{safe_tool}"
+
+
 def _convert_mcp_schema(server_name: str, mcp_tool) -> dict:
     """Convert an MCP tool listing to the Hermes registry schema format.
 
@@ -3689,9 +3709,7 @@ def _convert_mcp_schema(server_name: str, mcp_tool) -> dict:
     Returns:
         A dict suitable for ``registry.register(schema=...)``.
     """
-    safe_tool_name = sanitize_mcp_name_component(mcp_tool.name)
-    safe_server_name = sanitize_mcp_name_component(server_name)
-    prefixed_name = f"mcp_{safe_server_name}_{safe_tool_name}"
+    prefixed_name = mcp_prefixed_tool_name(server_name, mcp_tool.name)
     return {
         "name": prefixed_name,
         "description": mcp_tool.description or f"MCP tool {mcp_tool.name} from {server_name}",
@@ -3705,11 +3723,10 @@ def _build_utility_schemas(server_name: str) -> List[dict]:
     Returns a list of (schema, handler_factory_name) tuples encoded as dicts
     with keys: schema, handler_key.
     """
-    safe_name = sanitize_mcp_name_component(server_name)
     return [
         {
             "schema": {
-                "name": f"mcp_{safe_name}_list_resources",
+                "name": mcp_prefixed_tool_name(server_name, "list_resources"),
                 "description": f"List available resources from MCP server '{server_name}'",
                 "parameters": {
                     "type": "object",
@@ -3720,7 +3737,7 @@ def _build_utility_schemas(server_name: str) -> List[dict]:
         },
         {
             "schema": {
-                "name": f"mcp_{safe_name}_read_resource",
+                "name": mcp_prefixed_tool_name(server_name, "read_resource"),
                 "description": f"Read a resource by URI from MCP server '{server_name}'",
                 "parameters": {
                     "type": "object",
@@ -3737,7 +3754,7 @@ def _build_utility_schemas(server_name: str) -> List[dict]:
         },
         {
             "schema": {
-                "name": f"mcp_{safe_name}_list_prompts",
+                "name": mcp_prefixed_tool_name(server_name, "list_prompts"),
                 "description": f"List available prompts from MCP server '{server_name}'",
                 "parameters": {
                     "type": "object",
@@ -3748,7 +3765,7 @@ def _build_utility_schemas(server_name: str) -> List[dict]:
         },
         {
             "schema": {
-                "name": f"mcp_{safe_name}_get_prompt",
+                "name": mcp_prefixed_tool_name(server_name, "get_prompt"),
                 "description": f"Get a prompt by name from MCP server '{server_name}'",
                 "parameters": {
                     "type": "object",
@@ -4208,15 +4225,15 @@ def discover_mcp_tools() -> List[str]:
 def is_mcp_tool_parallel_safe(tool_name: str) -> bool:
     """Check if an MCP tool belongs to a server that supports parallel tool calls.
 
-    MCP tool names follow the pattern ``mcp_{server}_{tool}``, but that string
-    shape is ambiguous when server names contain underscores. Use the exact
-    server provenance captured at registration time rather than prefix
+    MCP tool names follow the pattern ``mcp__{server}__{tool}``, but that
+    string shape is ambiguous when server names contain underscores. Use the
+    exact server provenance captured at registration time rather than prefix
     matching, then check whether that server's config includes
     ``supports_parallel_tool_calls: true``.
 
     Returns False for non-MCP tools or tools from servers without the flag.
     """
-    if not tool_name.startswith("mcp_"):
+    if not tool_name.startswith(MCP_TOOL_NAME_PREFIX):
         return False
     with _lock:
         server_name = _mcp_tool_server_names.get(tool_name)