officially add sources

src/wf_mcp/broker/service/builtins.py

@@ -1,14 +1,57 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Protocol
 
-from wf_authoring import NodeSpec, node, runtime_error
+from pydantic import BaseModel, Field
 
+from wf_authoring import NodeReturn, NodeSpec, node, runtime_error
+
+from .sources import SpecSource
 from .specs import qualify_spec
 
 BUILTIN_CONNECTION_ID = "wf.std"
 """Internal source id for workflow standard-library node specs."""
 
+MCP_SOURCE_ID = "wf.mcp"
+"""Internal source id for broker MCP utility node specs."""
+
+
+class ToolCaller(Protocol):
+    """Small service boundary needed by the broker-local MCP utility nodes."""
+
+    async def call_tool(
+        self,
+        connection_id: str,
+        tool_name: str,
+        *,
+        arguments: dict[str, Any] | None = None,
+    ) -> dict[str, Any]: ...
+
+
+class McpCallToolInput(BaseModel):
+    """Input for calling a proxied MCP tool from inside a workflow."""
+
+    connection_id: str = Field(description="Connection id that owns the MCP tool.")
+    tool_name: str = Field(description="Local tool name on the upstream MCP server.")
+    arguments: dict[str, Any] = Field(
+        default_factory=dict,
+        description="JSON-compatible arguments passed to the upstream tool.",
+    )
+
+
+class McpCallToolOutput(BaseModel):
+    """Normalized output returned by a proxied MCP tool call."""
+
+    outcome: str = Field(description="Workflow outcome reported by the upstream tool.")
+    output: dict[str, Any] = Field(
+        default_factory=dict,
+        description="JSON-compatible tool result payload.",
+    )
+    meta: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Adapter metadata returned with the tool result.",
+    )
+
 
 def builtin_specs() -> dict[str, NodeSpec[Any, Any]]:
     """Return built-in NodeSpecs available to raw broker workflow plans."""
@@ -21,3 +64,44 @@ def builtin_specs() -> dict[str, NodeSpec[Any, Any]]:
     ]
     qualified_specs = [qualify_spec(BUILTIN_CONNECTION_ID, spec) for spec in specs]
     return {spec.name: spec for spec in qualified_specs}
+
+
+def mcp_specs(service: ToolCaller) -> dict[str, NodeSpec[Any, Any]]:
+    """Return service-bound MCP utility specs available to raw plans."""
+
+    @node(
+        name="call_tool",
+        outcomes=("ok", "error"),
+        input_model=McpCallToolInput,
+        output_model=McpCallToolOutput,
+        description="Call a tool on a registered MCP connection.",
+    )
+    async def call_tool(payload: McpCallToolInput) -> NodeReturn[McpCallToolOutput]:
+        result = await service.call_tool(
+            payload.connection_id,
+            payload.tool_name,
+            arguments=payload.arguments,
+        )
+        output = McpCallToolOutput.model_validate(result)
+        return NodeReturn(outcome=output.outcome, output=output)
+
+    qualified_specs = [qualify_spec(MCP_SOURCE_ID, call_tool)]
+    return {spec.name: spec for spec in qualified_specs}
+
+
+def builtin_sources(service: ToolCaller) -> dict[str, SpecSource]:
+    """Return all broker-local spec sources."""
+    return {
+        BUILTIN_CONNECTION_ID: SpecSource(
+            id=BUILTIN_CONNECTION_ID,
+            kind="system",
+            specs=builtin_specs(),
+            description="Workflow standard-library nodes.",
+        ),
+        MCP_SOURCE_ID: SpecSource(
+            id=MCP_SOURCE_ID,
+            kind="system",
+            specs=mcp_specs(service),
+            description="Broker MCP utility nodes.",
+        ),
+    }

src/wf_mcp/broker/service/core.py

@@ -10,6 +10,7 @@ from wf_core import NodeUse, Workflow, execute_workflow_async
 from ...connections import ConnectionRegistry, parse_connection_id, qualify_node_name
 from ...models import (
     AuthRecord,
+    CatalogNodeEntry,
     CatalogPromptEntry,
     CatalogResourceEntry,
     CatalogSnapshot,
@@ -23,7 +24,8 @@ from ..catalog import CombinedCatalog, snapshot_from_specs
 from ..discovery import discover_connection_capabilities, specs_from_discovered_tools
 from ..events import McpEvent, make_event
 from .adapters import require_adapter
-from .builtins import BUILTIN_CONNECTION_ID, builtin_specs
+from .builtins import builtin_sources
+from .sources import SpecSource
 from .specs import get_qualified_spec, qualify_spec
 
 
@@ -33,16 +35,20 @@ class WfMcpService:
     default_catalog_max_age_seconds: int = 300
     connections: ConnectionRegistry = field(default_factory=ConnectionRegistry)
     adapters: dict[str, BackendAdapter] = field(default_factory=dict)
-    specs_by_connection: dict[str, dict[str, NodeSpec[Any, Any]]] = field(
-        default_factory=dict
-    )
+    spec_sources: dict[str, SpecSource] = field(default_factory=dict)
     events: list[McpEvent] = field(default_factory=list)
     include_builtin_specs: bool = True
 
     def __post_init__(self) -> None:
-        """Install broker-local standard-library specs when enabled."""
+        """Install broker-local system specs when enabled."""
         if self.include_builtin_specs:
-            self.specs_by_connection.setdefault(BUILTIN_CONNECTION_ID, builtin_specs())
+            for source in builtin_sources(self).values():
+                self.register_spec_source(source)
+
+    @property
+    def specs_by_connection(self) -> dict[str, dict[str, NodeSpec[Any, Any]]]:
+        """Compatibility view of source specs keyed by source id."""
+        return {source.id: source.specs for source in self.spec_sources.values()}
 
     def register_connection(self, connection: ConnectionConfig) -> None:
         parse_connection_id(connection.id)
@@ -84,7 +90,14 @@ class WfMcpService:
             )
             for spec in specs
         }
-        self.specs_by_connection[connection_id] = qualified_specs
+        self.register_spec_source(
+            SpecSource(
+                id=connection_id,
+                kind="connection",
+                specs=qualified_specs,
+                description=f"Specs discovered or registered for {connection_id}.",
+            )
+        )
         snapshot = snapshot_from_specs(
             connection_id,
             specs=qualified_specs,
@@ -108,6 +121,39 @@ class WfMcpService:
                 snapshots[connection.id] = snapshot
         return CombinedCatalog(snapshots=snapshots)
 
+    def get_planner_catalog(self) -> CombinedCatalog:
+        """Return all planner-visible specs, including broker-local sources."""
+        snapshots = dict(self.get_catalog().snapshots)
+        fetched_at_epoch_ms = int(time.time() * 1000)
+        for source in self.spec_sources.values():
+            if not source.visible or source.kind != "system":
+                continue
+            snapshots[source.id] = snapshot_from_specs(
+                source.id,
+                specs=source.specs,
+                metadata={
+                    "kind": source.kind,
+                    "description": source.description,
+                },
+                fetched_at_epoch_ms=fetched_at_epoch_ms,
+                max_age_seconds=self.default_catalog_max_age_seconds,
+            )
+        return CombinedCatalog(snapshots=snapshots)
+
+    def list_spec_sources(self) -> list[dict[str, Any]]:
+        """Return planner spec sources without expanding every node schema."""
+        return [
+            source.as_status()
+            for source in sorted(
+                self.spec_sources.values(),
+                key=lambda source: source.id,
+            )
+        ]
+
+    def list_available_specs(self) -> list[CatalogNodeEntry]:
+        """Return planner-visible node catalog entries from every visible source."""
+        return self.get_planner_catalog().entries()
+
     def get_connection_snapshot(self, connection_id: str) -> CatalogSnapshot | None:
         self.connections.get(connection_id)
         return self.store.load_catalog(connection_id)
@@ -361,7 +407,10 @@ class WfMcpService:
             )
             snapshot = snapshot_from_specs(
                 connection_id,
-                specs=self.specs_by_connection.get(connection_id, {}),
+                specs=self.spec_sources.get(
+                    connection_id,
+                    SpecSource(id=connection_id, kind="connection"),
+                ).specs,
                 tool_display_names={
                     tool.name: tool.title for tool in capabilities.tools
                 },
@@ -446,8 +495,12 @@ class WfMcpService:
     def list_events(self) -> list[McpEvent]:
         return list(self.events)
 
+    def register_spec_source(self, source: SpecSource) -> None:
+        """Register a planner source."""
+        self.spec_sources[source.id] = source
+
     def _get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:
-        return get_qualified_spec(self.specs_by_connection, qualified_name)
+        return get_qualified_spec(self.spec_sources, qualified_name)
 
     def _record_event(self, event: McpEvent) -> None:
         self.events.append(event)

src/wf_mcp/broker/service/sources.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from wf_authoring import NodeSpec
+
+SpecSourceKind = Literal["connection", "system"]
+
+
+@dataclass(slots=True)
+class SpecSource:
+    """Planner-visible collection of workflow node specs.
+
+    Connection sources come from proxied MCP servers. System sources are local broker
+    capabilities, such as workflow stdlib nodes or service-bound MCP control nodes.
+    """
+
+    id: str
+    kind: SpecSourceKind
+    specs: dict[str, NodeSpec[Any, Any]] = field(default_factory=dict)
+    visible: bool = True
+    description: str | None = None
+
+    def as_status(self) -> dict[str, Any]:
+        """Return a compact payload suitable for UI and debugging surfaces."""
+        return {
+            "id": self.id,
+            "kind": self.kind,
+            "visible": self.visible,
+            "description": self.description,
+            "spec_count": len(self.specs),
+        }

src/wf_mcp/broker/service/specs.py

@@ -2,9 +2,12 @@ from __future__ import annotations
 
 from typing import Any
 
+from collections.abc import Mapping
+
 from wf_authoring import NodeSpec
 
 from ...connections import qualify_node_name
+from .sources import SpecSource
 
 
 def qualify_spec(connection_id: str, spec: NodeSpec[Any, Any]) -> NodeSpec[Any, Any]:
@@ -24,12 +27,12 @@ def qualify_spec(connection_id: str, spec: NodeSpec[Any, Any]) -> NodeSpec[Any,
 
 
 def get_qualified_spec(
-    specs_by_connection: dict[str, dict[str, NodeSpec[Any, Any]]],
+    spec_sources: Mapping[str, SpecSource],
     qualified_name: str,
 ) -> NodeSpec[Any, Any]:
-    """Resolve a namespaced node spec from the service's connection cache."""
-    connection_id, _ = qualified_name.rsplit(".", 1)
-    specs = specs_by_connection.get(connection_id)
-    if specs is None or qualified_name not in specs:
+    """Resolve a namespaced node spec from planner-visible sources."""
+    source_id, _ = qualified_name.rsplit(".", 1)
+    source = spec_sources.get(source_id)
+    if source is None or qualified_name not in source.specs:
         raise KeyError(f"unknown qualified node {qualified_name!r}")
-    return specs[qualified_name]
+    return source.specs[qualified_name]

src/wf_mcp/broker/tools.py

@@ -51,6 +51,14 @@ def register_broker_tools(server: FastMCP, service: WfMcpService) -> None:
     async def get_catalog() -> dict[str, Any]:
         return service.get_catalog().as_payload()
 
+    @server.tool()
+    async def get_planner_catalog() -> dict[str, Any]:
+        return service.get_planner_catalog().as_payload()
+
+    @server.tool()
+    async def list_spec_sources() -> list[dict[str, Any]]:
+        return service.list_spec_sources()
+
     @server.tool()
     async def read_broker_resource(qualified_name: str) -> dict[str, Any]:
         return await service.read_resource(qualified_name)