wf-mcp reorg 4

src/wf_mcp/broker/__init__.py

@@ -6,12 +6,12 @@ from .discovery import (
 )
 from .events import McpEvent, make_event
 from .server import (
-    build_service_from_config,
     create_broker_server,
-    load_broker_config,
     run_broker_server,
     run_transparent_proxy_server,
 )
+from .config import build_service_from_config, load_broker_config
+from .transport import normalize_transport
 from .service import WfMcpService
 
 __all__ = [
@@ -28,4 +28,5 @@ __all__ = [
     "run_transparent_proxy_server",
     "snapshot_from_specs",
     "specs_from_discovered_tools",
+    "normalize_transport",
 ]

src/wf_mcp/broker/config.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from ..control import BrokerConfigFile
+from ..models import BrokerConfig
+from ..sdk import McpSdkAdapter
+from ..storage import FileStore
+from .service import WfMcpService
+
+
+def load_broker_config(path: str | Path) -> BrokerConfig:
+    """Load a file-backed broker config into runtime config objects."""
+    config_path = Path(path)
+    data = json.loads(config_path.read_text(encoding="utf-8"))
+    return BrokerConfigFile.model_validate(data).to_runtime(config_path=config_path)
+
+
+def build_service_from_config(config: BrokerConfig) -> WfMcpService:
+    """Create a broker service with SDK adapters for configured connections."""
+    service = WfMcpService(store=FileStore(config.store_root))
+    for connection in config.connections:
+        service.register_connection(connection)
+        if connection.server not in service.adapters:
+            service.register_adapter(connection.server, McpSdkAdapter())
+    return service

src/wf_mcp/broker/prompts.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+import json
+
+from mcp.server.fastmcp import FastMCP
+
+from .service import WfMcpService
+
+
+def register_broker_prompts(server: FastMCP, service: WfMcpService) -> None:
+    """Register broker prompt handlers on a FastMCP server."""
+
+    @server.prompt(
+        name="plan_with_catalog",
+        description="Provide the broker catalog as planning context.",
+    )
+    def plan_with_catalog() -> list[dict[str, str]]:
+        payload = json.dumps(service.get_catalog().as_payload(), indent=2)
+        return [
+            {
+                "role": "user",
+                "content": (
+                    "Plan a workflow using this broker catalog. "
+                    "Prefer existing namespaced capabilities.\n\n"
+                    f"{payload}"
+                ),
+            }
+        ]

src/wf_mcp/broker/resources.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import json
+from dataclasses import asdict
+
+from mcp.server.fastmcp import FastMCP
+
+from .service import WfMcpService
+
+
+def register_broker_resources(server: FastMCP, service: WfMcpService) -> None:
+    """Register broker resource handlers on a FastMCP server."""
+
+    @server.resource("wf-mcp://catalog", name="catalog.all")
+    def catalog_resource() -> str:
+        return json.dumps(service.get_catalog().as_payload(), indent=2)
+
+    @server.resource(
+        "wf-mcp://connection/{connection_id}/catalog",
+        name="catalog.connection",
+    )
+    def connection_catalog_resource(connection_id: str) -> str:
+        snapshot = service.get_connection_snapshot(connection_id)
+        if snapshot is None:
+            raise KeyError(connection_id)
+        return json.dumps(
+            {
+                "connection_id": snapshot.connection_id,
+                "fetched_at_epoch_ms": snapshot.fetched_at_epoch_ms,
+                "max_age_seconds": snapshot.max_age_seconds,
+                "nodes": [asdict(node) for node in snapshot.nodes],
+                "resources": [asdict(resource) for resource in snapshot.resources],
+                "prompts": [asdict(prompt) for prompt in snapshot.prompts],
+                "metadata": snapshot.metadata,
+            },
+            indent=2,
+        )
+
+    @server.resource("wf-mcp://events", name="events.all")
+    def events_resource() -> str:
+        return json.dumps([asdict(event) for event in service.list_events()], indent=2)
+
+    @server.resource("wf-mcp://status", name="status.all")
+    def status_resource() -> str:
+        return json.dumps(service.connection_statuses(), indent=2)

src/wf_mcp/broker/server.py

@@ -1,35 +1,17 @@
 from __future__ import annotations
 
-import json
 import os
-from dataclasses import asdict
 from pathlib import Path
-from typing import Any, Literal
 
 from mcp.server.fastmcp import FastMCP
 
-from ..control import BrokerConfigFile
-from ..models import BrokerConfig
-from ..sdk import McpSdkAdapter
-from ..shared.errors import error_payload
-from ..storage import FileStore
 from ..transparent_proxy import create_transparent_proxy_server
+from .config import build_service_from_config, load_broker_config
+from .prompts import register_broker_prompts
+from .resources import register_broker_resources
 from .service import WfMcpService
-
+from .tools import register_broker_tools
-
+from .transport import normalize_transport
-def load_broker_config(path: str | Path) -> BrokerConfig:
-    config_path = Path(path)
-    data = json.loads(config_path.read_text(encoding="utf-8"))
-    return BrokerConfigFile.model_validate(data).to_runtime(config_path=config_path)
-
-
-def build_service_from_config(config: BrokerConfig) -> WfMcpService:
-    service = WfMcpService(store=FileStore(config.store_root))
-    for connection in config.connections:
-        service.register_connection(connection)
-        if connection.server not in service.adapters:
-            service.register_adapter(connection.server, McpSdkAdapter())
-    return service
 
 
 def create_broker_server(service: WfMcpService) -> FastMCP:
@@ -42,151 +24,9 @@ def create_broker_server(service: WfMcpService) -> FastMCP:
         ),
     )
 
-    @server.tool()
+    register_broker_tools(server, service)
-    async def list_connections() -> list[dict[str, Any]]:
+    register_broker_resources(server, service)
-        return [
+    register_broker_prompts(server, service)
-            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()
-
-    @server.tool()
-    async def refresh_connection_catalog(connection_id: str) -> dict[str, Any]:
-        try:
-            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()
-
-    @server.tool()
-    async def read_broker_resource(qualified_name: str) -> dict[str, Any]:
-        return await service.read_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)
-
-    @server.tool()
-    async def invoke_broker_method(
-        connection_id: str,
-        method: str,
-        params: dict[str, Any] | None = None,
-    ) -> dict[str, Any]:
-        try:
-            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(
-        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 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()]
-
-    @server.resource("wf-mcp://catalog", name="catalog.all")
-    def catalog_resource() -> str:
-        return json.dumps(service.get_catalog().as_payload(), indent=2)
-
-    @server.resource(
-        "wf-mcp://connection/{connection_id}/catalog",
-        name="catalog.connection",
-    )
-    def connection_catalog_resource(connection_id: str) -> str:
-        snapshot = service.get_connection_snapshot(connection_id)
-        if snapshot is None:
-            raise KeyError(connection_id)
-        return json.dumps(
-            {
-                "connection_id": snapshot.connection_id,
-                "fetched_at_epoch_ms": snapshot.fetched_at_epoch_ms,
-                "max_age_seconds": snapshot.max_age_seconds,
-                "nodes": [asdict(node) for node in snapshot.nodes],
-                "resources": [asdict(resource) for resource in snapshot.resources],
-                "prompts": [asdict(prompt) for prompt in snapshot.prompts],
-                "metadata": snapshot.metadata,
-            },
-            indent=2,
-        )
-
-    @server.resource("wf-mcp://events", name="events.all")
-    def events_resource() -> str:
-        return json.dumps([asdict(event) for event in service.list_events()], indent=2)
-
-    @server.resource("wf-mcp://status", name="status.all")
-    def status_resource() -> str:
-        return json.dumps(service.connection_statuses(), indent=2)
-
-    @server.prompt(
-        name="plan_with_catalog",
-        description="Provide the broker catalog as planning context.",
-    )
-    def plan_with_catalog() -> list[dict[str, str]]:
-        payload = json.dumps(service.get_catalog().as_payload(), indent=2)
-        return [
-            {
-                "role": "user",
-                "content": (
-                    "Plan a workflow using this broker catalog. "
-                    "Prefer existing namespaced capabilities.\n\n"
-                    f"{payload}"
-                ),
-            }
-        ]
-
     return server
 
 
@@ -196,20 +36,6 @@ def main() -> None:
     run_broker_server(config_path, transport_env)
 
 
-def normalize_transport(
-    transport: str,
-) -> Literal["stdio", "sse", "streamable-http"]:
-    match transport:
-        case "streamable_http" | "streamable-http":
-            return "streamable-http"
-        case "stdio":
-            return "stdio"
-        case "sse":
-            return "sse"
-        case _:
-            raise ValueError(f"we dont support {transport} yet sry")
-
-
 def run_broker_server(config_path: str | Path, transport: str = "stdio") -> None:
     config = load_broker_config(config_path)
     service = build_service_from_config(config)

src/wf_mcp/broker/tools.py

@@ -0,0 +1,108 @@
+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 .service import WfMcpService
+
+
+def register_broker_tools(server: FastMCP, service: WfMcpService) -> None:
+    """Register broker tool handlers on a FastMCP server."""
+
+    @server.tool()
+    async def list_connections() -> list[dict[str, Any]]:
+        return [
+            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()
+
+    @server.tool()
+    async def refresh_connection_catalog(connection_id: str) -> dict[str, Any]:
+        try:
+            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()
+
+    @server.tool()
+    async def read_broker_resource(qualified_name: str) -> dict[str, Any]:
+        return await service.read_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)
+
+    @server.tool()
+    async def invoke_broker_method(
+        connection_id: str,
+        method: str,
+        params: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        try:
+            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(
+        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 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()]

src/wf_mcp/broker/transport.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from typing import Literal
+
+
+def normalize_transport(
+    transport: str,
+) -> Literal["stdio", "sse", "streamable-http"]:
+    """Normalize supported CLI transport spellings for FastMCP."""
+    match transport:
+        case "streamable_http" | "streamable-http":
+            return "streamable-http"
+        case "stdio":
+            return "stdio"
+        case "sse":
+            return "sse"
+        case _:
+            raise ValueError(f"we dont support {transport} yet sry")