docs enforcement; safe tool names for claude desktop / fragile ahh harnesses

tests/wf_mcp/test_cli.py

@@ -53,6 +53,7 @@ def test_build_parser_accepts_proxy_compatibility_flags() -> None:
             "--resources-as-tools",
             "--prompts-as-tools",
             "--search-tools",
+            "--safe-tool-names",
         ]
     )
 
@@ -60,6 +61,7 @@ def test_build_parser_accepts_proxy_compatibility_flags() -> None:
     assert args.resources_as_tools is True
     assert args.prompts_as_tools is True
     assert args.search_tools is True
+    assert args.safe_tool_names is True
 
 
 def test_build_parser_rejects_legacy_mode_flag() -> None:

tests/wf_mcp/test_safe_tool_names.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import asyncio
+
+from fastmcp import FastMCP
+
+from wf_mcp.transparent_proxy.safe_names import (
+    SafeToolNames,
+    encode_safe_tool_name,
+)
+
+
+def test_encode_safe_tool_name_keeps_readable_names() -> None:
+    assert encode_safe_tool_name("wf.workflow.list_artifacts") == (
+        "wf_workflow_list_artifacts"
+    )
+    assert encode_safe_tool_name("search_tools") == "search_tools"
+    assert encode_safe_tool_name("some-tool") == "some-tool"
+
+
+def test_safe_tool_names_hashes_collisions_and_preserves_lookup_invariants() -> None:
+    transform = SafeToolNames()
+    server = _server_with_tools("demo.echo", "demo_echo", transform=transform)
+
+    tools = asyncio.run(server.list_tools())
+    names = [tool.name for tool in tools]
+
+    assert "demo_echo" in names
+    assert any(name.startswith("demo_echo_h") for name in names)
+    transform.assert_consistent()
+
+
+def test_safe_tool_names_hashes_overlength_names() -> None:
+    transform = SafeToolNames()
+    server = _server_with_tools("x" * 65, transform=transform)
+
+    tools = asyncio.run(server.list_tools())
+
+    assert len(tools[0].name) <= 64
+    assert "_h" in tools[0].name
+    transform.assert_consistent()
+
+
+def _server_with_tools(
+    *names: str,
+    transform: SafeToolNames | None = None,
+) -> FastMCP[object]:
+    server: FastMCP[object] = FastMCP("safe-name-test")
+    for name in names:
+
+        def handler() -> None:
+            return None
+
+        server.tool(name=name)(handler)
+    server.add_transform(transform or SafeToolNames())
+    return server

tests/wf_mcp/test_server.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+import re
 import sys
 from typing import Any
 
@@ -20,6 +21,23 @@ def _structured(result: Any) -> dict[str, Any]:
     return content
 
 
+async def _assert_safe_tool_maps(
+    client: Any,
+    *,
+    original_name: str,
+    safe_name: str,
+    arguments: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Assert one safe public tool name maps back to its original tool."""
+    tools = await client.list_tools()
+    names = [tool.name for tool in tools]
+    assert safe_name in names
+    assert original_name not in names
+    assert all(re.fullmatch(r"^[a-zA-Z0-9_-]{1,64}$", name) for name in names)
+    assert len(names) == len(set(names))
+    return _structured(await client.call_tool(safe_name, arguments or {}))
+
+
 def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
     config = BrokerConfig(
         store_root=local_temp_root() / "unified_server_store",
@@ -99,8 +117,8 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "error_message_source" in minimal_request["properties"]
             assert (
                 minimal_request["properties"]["input_schema"]["description"]
-                == "JSON Schema object. Keep this as ordinary JSON; "
-                "nested schema fields are passed through unchanged."
+                == "Public input JSON Schema for the workflow or wrapper being "
+                "drafted."
             )
             wrapper_workspace_input = tools_by_name[
                 "wf.workflow.create_wrapper_from_workspace"
@@ -232,6 +250,88 @@ def test_server_search_mode_pins_stable_control_and_workflow_tools() -> None:
     asyncio.run(run_proxy())
 
 
+def test_server_search_mode_can_use_safe_tool_names() -> None:
+    config = BrokerConfig(
+        store_root=local_temp_root() / "server_search_safe_names_store",
+        connections=[
+            ConnectionConfig(
+                id="fixture.personal",
+                server="fixture",
+                account="personal",
+                metadata={
+                    "transport": "stdio",
+                    "command": sys.executable,
+                    "args": [fixture_server_path()],
+                },
+            )
+        ],
+    )
+
+    async def run_proxy() -> None:
+        client = create_server_client(
+            config,
+            search_tools=True,
+            safe_tool_names=True,
+        )
+        async with client:
+            tools = await client.list_tools()
+            names = [tool.name for tool in tools]
+
+            assert "search_tools" in names
+            assert "call_tool" in names
+            assert "wf_admin_list_sources" in names
+            assert "wf_workflow_call_capability" in names
+            assert "wf.admin.list_sources" not in names
+
+            result = await _assert_safe_tool_maps(
+                client,
+                original_name="wf.admin.list_sources",
+                safe_name="wf_admin_list_sources",
+            )
+            source_ids = {source["id"] for source in result["sources"]}
+            assert "wf.std" in source_ids
+
+    asyncio.run(run_proxy())
+
+
+def test_server_safe_tool_names_adapts_dotted_runtime_names() -> None:
+    config = BrokerConfig(
+        store_root=local_temp_root() / "server_safe_tool_names_store",
+        connections=[
+            ConnectionConfig(
+                id="fixture.personal",
+                server="fixture",
+                account="personal",
+                metadata={
+                    "transport": "stdio",
+                    "command": sys.executable,
+                    "args": [fixture_server_path()],
+                },
+            )
+        ],
+    )
+
+    async def run_proxy() -> None:
+        client = create_server_client(config, safe_tool_names=True)
+        async with client:
+            artifacts = await _assert_safe_tool_maps(
+                client,
+                original_name="wf.workflow.list_artifacts",
+                safe_name="wf_workflow_list_artifacts",
+            )
+            echo = await _assert_safe_tool_maps(
+                client,
+                original_name="fixture.personal.echo_tool",
+                safe_name="fixture_personal_echo_tool",
+                arguments={"text": "hello"},
+            )
+
+            assert artifacts["nodes"] == []
+            assert echo["echoed"] == "hello"
+
+    asyncio.run(run_proxy())
+
+
 def test_workflow_tools_have_human_metadata() -> None:
     config = BrokerConfig(
         store_root=local_temp_root() / "unified_metadata_store",