more functions out of the Big Tool registers

docs/superpowers/plans/2026-05-12-unified-mcp-surface.md

@@ -244,10 +244,10 @@ config_reloaded
 
 - [ ] Move workflow artifact list/save/inspect/validate/run logic into shared
   handler functions/classes.
-- [ ] Move admin list/refresh/config/reload logic into shared handler
+- [x] Move admin list/refresh/config/reload logic into shared handler
   functions/classes.
-- [ ] Keep broker compatibility tool names working.
+- [x] Keep broker compatibility tool names working.
-- [ ] Keep transparent proxy admin tool names working.
+- [x] Keep transparent proxy admin tool names working.
 - [ ] Do not change behavior in this phase; only remove duplicated logic and
   create a single implementation path.
 

src/wf_mcp/admin_surface/__init__.py

@@ -0,0 +1,3 @@
+from .handlers import BrokerAdminHandlers, ProxyAdminRuntime, TransparentAdminHandlers
+
+__all__ = ["BrokerAdminHandlers", "ProxyAdminRuntime", "TransparentAdminHandlers"]

src/wf_mcp/admin_surface/handlers/__init__.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from .broker import BrokerAdminHandlers
+from .runtime import ProxyAdminRuntime
+from .transparent import TransparentAdminHandlers
+
+__all__ = [
+    "BrokerAdminHandlers",
+    "ProxyAdminRuntime",
+    "TransparentAdminHandlers",
+]

src/wf_mcp/admin_surface/handlers/broker.py

@@ -0,0 +1,107 @@
+from dataclasses import asdict
+from typing import Any
+
+from wf_mcp.broker.service import WfMcpService
+from wf_mcp.shared.errors import error_payload
+
+
+class BrokerAdminHandlers:
+    """Shared implementation for service-backed broker admin operations."""
+
+    def __init__(self, service: WfMcpService) -> None:
+        self.service = service
+
+    def list_connections(self) -> list[dict[str, Any]]:
+        return [
+            asdict(connection)
+            for connection in sorted(
+                self.service.connections.list_all(),
+                key=lambda connection: connection.id,
+            )
+        ]
+
+    def get_connection_statuses(self) -> list[dict[str, Any]]:
+        return self.service.connection_statuses()
+
+    async def refresh_connection_catalog(self, connection_id: str) -> dict[str, Any]:
+        try:
+            await self.service.refresh_connection_catalog(connection_id)
+        except Exception as exc:
+            return {
+                "connection_id": connection_id,
+                "refreshed": False,
+                **error_payload(exc),
+            }
+        snapshot = self.service.get_connection_snapshot(connection_id)
+        if snapshot is None:
+            return {"connection_id": connection_id, "refreshed": False}
+        return {
+            "connection_id": connection_id,
+            "refreshed": True,
+            "node_count": len(snapshot.nodes),
+            "resource_count": len(snapshot.resources),
+            "prompt_count": len(snapshot.prompts),
+        }
+
+    def get_catalog(self) -> dict[str, Any]:
+        return self.service.get_catalog().as_payload()
+
+    def get_planner_catalog(self) -> dict[str, Any]:
+        return self.service.get_planner_catalog().as_payload()
+
+    def list_spec_sources(self) -> list[dict[str, Any]]:
+        return self.service.list_spec_sources()
+
+    async def read_broker_resource(self, qualified_name: str) -> dict[str, Any]:
+        return await self.service.read_resource(qualified_name)
+
+    async def render_broker_prompt(
+        self,
+        qualified_name: str,
+        arguments: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        return await self.service.render_prompt(qualified_name, arguments=arguments)
+
+    async def invoke_broker_method(
+        self,
+        connection_id: str,
+        method: str,
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        try:
+            return await self.service.invoke_method(connection_id, method, params=params)
+        except Exception as exc:
+            return {
+                "connection_id": connection_id,
+                "method": method,
+                "ok": False,
+                **error_payload(exc),
+            }
+
+    async def call_broker_tool(
+        self,
+        connection_id: str,
+        tool_name: str,
+        arguments: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        try:
+            return {
+                "connection_id": connection_id,
+                "tool_name": tool_name,
+                "ok": True,
+                **await self.service.call_tool(
+                    connection_id,
+                    tool_name,
+                    arguments=arguments,
+                ),
+            }
+        except Exception as exc:
+            return {
+                "connection_id": connection_id,
+                "tool_name": tool_name,
+                "ok": False,
+                **error_payload(exc),
+            }
+
+    def get_broker_events(self) -> list[dict[str, Any]]:
+        return [asdict(event) for event in self.service.list_events()]

src/wf_mcp/admin_surface/handlers/config.py

@@ -0,0 +1,36 @@
+from typing import Any, Protocol
+
+
+class ConfigManager(Protocol):
+    """Config mutation methods used by transparent admin handlers."""
+
+    def get_payload(self) -> dict[str, Any]: ...
+
+    def add_connection(
+        self,
+        *,
+        connection_id: str,
+        server: str,
+        account: str,
+        metadata: dict[str, Any] | None = None,
+        enabled: bool = True,
+    ) -> dict[str, Any]: ...
+
+    def update_connection(
+        self,
+        *,
+        connection_id: str,
+        server: str | None = None,
+        account: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        enabled: bool | None = None,
+    ) -> dict[str, Any]: ...
+
+    def set_connection_enabled(
+        self,
+        connection_id: str,
+        *,
+        enabled: bool,
+    ) -> dict[str, Any]: ...
+
+    def remove_connection(self, connection_id: str) -> dict[str, Any]: ...

src/wf_mcp/admin_surface/handlers/runtime.py

@@ -0,0 +1,28 @@
+from typing import Any, Protocol
+
+from wf_mcp.admin_surface.handlers.config import ConfigManager
+from wf_mcp.models import BrokerConfig
+
+
+class ProxyAdminRuntime(Protocol):
+    """Runtime boundary needed by transparent proxy admin handlers."""
+
+    @property
+    def manager(self) -> ConfigManager | None: ...
+
+    def current_config(self) -> BrokerConfig: ...
+
+    def require_manager(self) -> ConfigManager: ...
+
+    def reload(self) -> dict[str, Any]: ...
+
+    async def list_proxy_tools_page(
+        self,
+        *,
+        connection_id: str | None = None,
+        query: str | None = None,
+        limit: int = 50,
+        cursor: str | None = None,
+    ) -> dict[str, Any]: ...
+
+    async def get_proxy_tool(self, proxy_name: str) -> dict[str, Any]: ...

src/wf_mcp/admin_surface/handlers/transparent.py

@@ -0,0 +1,111 @@
+from dataclasses import asdict
+from typing import Any
+
+from .runtime import ProxyAdminRuntime
+
+
+class TransparentAdminHandlers:
+    """Shared implementation for transparent-proxy config/admin operations."""
+
+    def __init__(self, runtime: ProxyAdminRuntime) -> None:
+        self.runtime = runtime
+
+    def list_connections(self) -> list[dict[str, Any]]:
+        return [
+            asdict(connection)
+            for connection in sorted(
+                self.runtime.current_config().connections,
+                key=lambda connection: connection.id,
+            )
+        ]
+
+    def get_connection_statuses(self) -> list[dict[str, Any]]:
+        return [
+            {
+                "connection_id": connection.id,
+                "server": connection.server,
+                "account": connection.account,
+                "enabled": connection.enabled,
+                "transport": connection.metadata.get("transport"),
+            }
+            for connection in sorted(
+                self.runtime.current_config().connections,
+                key=lambda connection: connection.id,
+            )
+        ]
+
+    def get_config(self) -> dict[str, Any]:
+        if self.runtime.manager is not None:
+            return self.runtime.manager.get_payload()
+        config = self.runtime.current_config()
+        return {
+            "store_root": str(config.store_root),
+            "connections": [asdict(connection) for connection in config.connections],
+        }
+
+    def reload_config(self) -> dict[str, Any]:
+        return self.runtime.reload()
+
+    async def list_proxy_tools(
+        self,
+        connection_id: str | None = None,
+        query: str | None = None,
+        limit: int = 50,
+        cursor: str | None = None,
+    ) -> dict[str, Any]:
+        return await self.runtime.list_proxy_tools_page(
+            connection_id=connection_id,
+            query=query,
+            limit=limit,
+            cursor=cursor,
+        )
+
+    async def get_proxy_tool(self, proxy_name: str) -> dict[str, Any]:
+        return await self.runtime.get_proxy_tool(proxy_name)
+
+    def add_connection(
+        self,
+        connection_id: str,
+        server: str,
+        account: str,
+        metadata: dict[str, Any] | None = None,
+        enabled: bool = True,
+    ) -> dict[str, Any]:
+        return self.runtime.require_manager().add_connection(
+            connection_id=connection_id,
+            server=server,
+            account=account,
+            metadata=metadata,
+            enabled=enabled,
+        )
+
+    def update_connection(
+        self,
+        connection_id: str,
+        server: str | None = None,
+        account: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        enabled: bool | None = None,
+    ) -> dict[str, Any]:
+        return self.runtime.require_manager().update_connection(
+            connection_id=connection_id,
+            server=server,
+            account=account,
+            metadata=metadata,
+            enabled=enabled,
+        )
+
+    def enable_connection(self, connection_id: str) -> dict[str, Any]:
+        return self.runtime.require_manager().set_connection_enabled(
+            connection_id,
+            enabled=True,
+        )
+
+    def disable_connection(self, connection_id: str) -> dict[str, Any]:
+        return self.runtime.require_manager().set_connection_enabled(
+            connection_id,
+            enabled=False,
+        )
+
+    def remove_connection(self, connection_id: str) -> dict[str, Any]:
+        return self.runtime.require_manager().remove_connection(connection_id)

src/wf_mcp/broker/tools.py

@@ -1,77 +1,54 @@
 from __future__ import annotations
 
-from dataclasses import asdict
 from typing import Any
 
 from mcp.server.fastmcp import FastMCP
 
-from ..shared.errors import error_payload
+from ..admin_surface import BrokerAdminHandlers
 from .service import WfMcpService
 
 
 def register_broker_tools(server: FastMCP, service: WfMcpService) -> None:
     """Register broker tool handlers on a FastMCP server."""
+    handlers = BrokerAdminHandlers(service)
 
     # These MCP tool names are compatibility exports. Their capability metadata
     # belongs to the wf.admin source; future admin-enabled servers can project
     # dotted wf.admin.* names from that source.
     @server.tool()
     async def list_connections() -> list[dict[str, Any]]:
-        return [
+        return handlers.list_connections()
-            asdict(connection)
-            for connection in sorted(
-                service.connections.list_all(),
-                key=lambda connection: connection.id,
-            )
-        ]
 
     @server.tool()
     async def get_connection_statuses() -> list[dict[str, Any]]:
-        return service.connection_statuses()
+        return handlers.get_connection_statuses()
 
     @server.tool()
     async def refresh_connection_catalog(connection_id: str) -> dict[str, Any]:
-        try:
+        return await handlers.refresh_connection_catalog(connection_id)
-            await service.refresh_connection_catalog(connection_id)
-        except Exception as exc:
-            return {
-                "connection_id": connection_id,
-                "refreshed": False,
-                **error_payload(exc),
-            }
-        snapshot = service.get_connection_snapshot(connection_id)
-        if snapshot is None:
-            return {"connection_id": connection_id, "refreshed": False}
-        return {
-            "connection_id": connection_id,
-            "refreshed": True,
-            "node_count": len(snapshot.nodes),
-            "resource_count": len(snapshot.resources),
-            "prompt_count": len(snapshot.prompts),
-        }
 
     @server.tool()
     async def get_catalog() -> dict[str, Any]:
-        return service.get_catalog().as_payload()
+        return handlers.get_catalog()
 
     @server.tool()
     async def get_planner_catalog() -> dict[str, Any]:
-        return service.get_planner_catalog().as_payload()
+        return handlers.get_planner_catalog()
 
     @server.tool()
     async def list_spec_sources() -> list[dict[str, Any]]:
-        return service.list_spec_sources()
+        return handlers.list_spec_sources()
 
     @server.tool()
     async def read_broker_resource(qualified_name: str) -> dict[str, Any]:
-        return await service.read_resource(qualified_name)
+        return await handlers.read_broker_resource(qualified_name)
 
     @server.tool()
     async def render_broker_prompt(
         qualified_name: str,
         arguments: dict[str, str] | None = None,
     ) -> dict[str, Any]:
-        return await service.render_prompt(qualified_name, arguments=arguments)
+        return await handlers.render_broker_prompt(qualified_name, arguments=arguments)
 
     @server.tool()
     async def invoke_broker_method(
@@ -79,15 +56,7 @@ def register_broker_tools(server: FastMCP, service: WfMcpService) -> None:
         method: str,
         params: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
-        try:
+        return await handlers.invoke_broker_method(connection_id, method, params=params)
-            return await service.invoke_method(connection_id, method, params=params)
-        except Exception as exc:
-            return {
-                "connection_id": connection_id,
-                "method": method,
-                "ok": False,
-                **error_payload(exc),
-            }
 
     @server.tool()
     async def call_broker_tool(
@@ -95,25 +64,12 @@ def register_broker_tools(server: FastMCP, service: WfMcpService) -> None:
         tool_name: str,
         arguments: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
-        try:
+        return await handlers.call_broker_tool(
-            return {
-                "connection_id": connection_id,
-                "tool_name": tool_name,
-                "ok": True,
-                **await service.call_tool(
             connection_id,
             tool_name,
             arguments=arguments,
-                ),
+        )
-            }
-        except Exception as exc:
-            return {
-                "connection_id": connection_id,
-                "tool_name": tool_name,
-                "ok": False,
-                **error_payload(exc),
-            }
 
     @server.tool()
     async def get_broker_events() -> list[dict[str, Any]]:
-        return [asdict(event) for event in service.list_events()]
+        return handlers.get_broker_events()

src/wf_mcp/transparent_proxy/admin.py

@@ -1,33 +1,10 @@
 from __future__ import annotations
 
-from dataclasses import asdict
+from typing import Any
-from typing import Any, Protocol
 
 from fastmcp import FastMCP
 
-from ..control import BrokerConfigManager
+from ..admin_surface import ProxyAdminRuntime, TransparentAdminHandlers
-from ..models import BrokerConfig
-
-
-class ProxyAdminRuntime(Protocol):
-    manager: BrokerConfigManager | None
-
-    def current_config(self) -> BrokerConfig: ...
-
-    def require_manager(self) -> BrokerConfigManager: ...
-
-    def reload(self) -> dict[str, Any]: ...
-
-    async def list_proxy_tools_page(
-        self,
-        *,
-        connection_id: str | None = None,
-        query: str | None = None,
-        limit: int = 50,
-        cursor: str | None = None,
-    ) -> dict[str, Any]: ...
-
-    async def get_proxy_tool(self, proxy_name: str) -> dict[str, Any]: ...
 
 
 def create_proxy_admin_server(
@@ -38,46 +15,23 @@ def create_proxy_admin_server(
         "wf-mcp-admin",
         instructions="Administrative tools for this wf-mcp proxy instance.",
     )
+    handlers = TransparentAdminHandlers(runtime)
 
     @admin.tool()
     async def list_connections() -> list[dict[str, Any]]:
-        return [
+        return handlers.list_connections()
-            asdict(connection)
-            for connection in sorted(
-                runtime.current_config().connections,
-                key=lambda connection: connection.id,
-            )
-        ]
 
     @admin.tool()
     async def get_connection_statuses() -> list[dict[str, Any]]:
-        return [
+        return handlers.get_connection_statuses()
-            {
-                "connection_id": connection.id,
-                "server": connection.server,
-                "account": connection.account,
-                "enabled": connection.enabled,
-                "transport": connection.metadata.get("transport"),
-            }
-            for connection in sorted(
-                runtime.current_config().connections,
-                key=lambda connection: connection.id,
-            )
-        ]
 
     @admin.tool()
     async def get_config() -> dict[str, Any]:
-        if runtime.manager is not None:
+        return handlers.get_config()
-            return runtime.manager.get_payload()
-        config = runtime.current_config()
-        return {
-            "store_root": str(config.store_root),
-            "connections": [asdict(connection) for connection in config.connections],
-        }
 
     @admin.tool()
     async def reload_config() -> dict[str, Any]:
-        return runtime.reload()
+        return handlers.reload_config()
 
     @admin.tool()
     async def list_proxy_tools(
@@ -86,7 +40,7 @@ def create_proxy_admin_server(
         limit: int = 50,
         cursor: str | None = None,
     ) -> dict[str, Any]:
-        return await runtime.list_proxy_tools_page(
+        return await handlers.list_proxy_tools(
             connection_id=connection_id,
             query=query,
             limit=limit,
@@ -95,7 +49,7 @@ def create_proxy_admin_server(
 
     @admin.tool()
     async def get_proxy_tool(proxy_name: str) -> dict[str, Any]:
-        return await runtime.get_proxy_tool(proxy_name)
+        return await handlers.get_proxy_tool(proxy_name)
 
     @admin.tool()
     async def add_connection(
@@ -105,7 +59,7 @@ def create_proxy_admin_server(
         metadata: dict[str, Any] | None = None,
         enabled: bool = True,
     ) -> dict[str, Any]:
-        return runtime.require_manager().add_connection(
+        return handlers.add_connection(
             connection_id=connection_id,
             server=server,
             account=account,
@@ -121,7 +75,7 @@ def create_proxy_admin_server(
         metadata: dict[str, Any] | None = None,
         enabled: bool | None = None,
     ) -> dict[str, Any]:
-        return runtime.require_manager().update_connection(
+        return handlers.update_connection(
             connection_id=connection_id,
             server=server,
             account=account,
@@ -131,20 +85,14 @@ def create_proxy_admin_server(
 
     @admin.tool()
     async def enable_connection(connection_id: str) -> dict[str, Any]:
-        return runtime.require_manager().set_connection_enabled(
+        return handlers.enable_connection(connection_id)
-            connection_id,
-            enabled=True,
-        )
 
     @admin.tool()
     async def disable_connection(connection_id: str) -> dict[str, Any]:
-        return runtime.require_manager().set_connection_enabled(
+        return handlers.disable_connection(connection_id)
-            connection_id,
-            enabled=False,
-        )
 
     @admin.tool()
     async def remove_connection(connection_id: str) -> dict[str, Any]:
-        return runtime.require_manager().remove_connection(connection_id)
+        return handlers.remove_connection(connection_id)
 
     return admin

tests/wf_mcp/test_admin_surface.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from wf_mcp.admin_surface import BrokerAdminHandlers, TransparentAdminHandlers
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import BrokerConfig, ConnectionConfig
+from wf_mcp.storage import FileStore
+
+from .test_support import local_temp_root
+
+
+def test_broker_admin_handlers_list_connections_and_events() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "admin_broker_store"))
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    handlers = BrokerAdminHandlers(service)
+
+    connections = handlers.list_connections()
+    events = handlers.get_broker_events()
+
+    assert connections[0]["id"] == "demo.personal"
+    assert connections[0]["server"] == "demo"
+    assert events[0]["kind"] == "connection_registered"
+    assert events[0]["connection_id"] == "demo.personal"
+
+
+def test_broker_admin_handlers_report_failed_refresh_payload() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "admin_refresh_store"))
+    handlers = BrokerAdminHandlers(service)
+
+    payload = _run(handlers.refresh_connection_catalog("missing.personal"))
+
+    assert payload["connection_id"] == "missing.personal"
+    assert payload["refreshed"] is False
+    assert payload["error_type"] == "KeyError"
+
+
+def test_transparent_admin_handlers_delegate_config_operations() -> None:
+    runtime = FakeProxyAdminRuntime()
+    handlers = TransparentAdminHandlers(runtime)
+
+    connections = handlers.list_connections()
+    statuses = handlers.get_connection_statuses()
+    config = handlers.get_config()
+    add_payload = handlers.add_connection(
+        connection_id="demo.work",
+        server="demo",
+        account="work",
+    )
+
+    assert connections[0]["id"] == "demo.personal"
+    assert statuses[0]["transport"] == "stdio"
+    assert config["source"] == "manager"
+    assert add_payload["action"] == "add_connection"
+    assert runtime.manager.added[0]["connection_id"] == "demo.work"
+
+
+async def _await_value(value: Any) -> Any:
+    return await value
+
+
+def _run(value: Any) -> Any:
+    import asyncio
+
+    return asyncio.run(_await_value(value))
+
+
+@dataclass
+class FakeManager:
+    added: list[dict[str, Any]]
+
+    def get_payload(self) -> dict[str, Any]:
+        return {"source": "manager"}
+
+    def add_connection(
+        self,
+        *,
+        connection_id: str,
+        server: str,
+        account: str,
+        metadata: dict[str, Any] | None = None,
+        enabled: bool = True,
+    ) -> dict[str, Any]:
+        self.added.append(
+            {
+                "connection_id": connection_id,
+                "server": server,
+                "account": account,
+                "metadata": metadata,
+                "enabled": enabled,
+            }
+        )
+        return {"action": "add_connection", "ok": True}
+
+    def update_connection(
+        self,
+        *,
+        connection_id: str,
+        server: str | None = None,
+        account: str | None = None,
+        metadata: dict[str, Any] | None = None,
+        enabled: bool | None = None,
+    ) -> dict[str, Any]:
+        return {"action": "update_connection", "connection_id": connection_id}
+
+    def set_connection_enabled(
+        self,
+        connection_id: str,
+        *,
+        enabled: bool,
+    ) -> dict[str, Any]:
+        return {
+            "action": "update_connection",
+            "connection_id": connection_id,
+            "enabled": enabled,
+        }
+
+    def remove_connection(self, connection_id: str) -> dict[str, Any]:
+        return {"action": "remove_connection", "connection_id": connection_id}
+
+
+class FakeProxyAdminRuntime:
+    def __init__(self) -> None:
+        self.manager = FakeManager(added=[])
+        self._config = BrokerConfig(
+            store_root=local_temp_root() / "transparent_admin_handlers_store",
+            connections=[
+                ConnectionConfig(
+                    id="demo.personal",
+                    server="demo",
+                    account="personal",
+                    metadata={"transport": "stdio"},
+                )
+            ],
+        )
+
+    def current_config(self) -> BrokerConfig:
+        return self._config
+
+    def require_manager(self) -> FakeManager:
+        return self.manager
+
+    def reload(self) -> dict[str, Any]:
+        return {"ok": True, "reloaded": True}
+
+    async def list_proxy_tools_page(
+        self,
+        *,
+        connection_id: str | None = None,
+        query: str | None = None,
+        limit: int = 50,
+        cursor: str | None = None,
+    ) -> dict[str, Any]:
+        return {"tools": [], "nextCursor": None, "total": 0}
+
+    async def get_proxy_tool(self, proxy_name: str) -> dict[str, Any]:
+        return {"proxy_name": proxy_name}