new mcp tools

docs/workflow_capabilities.md

@@ -247,6 +247,20 @@ That direct-call surface is different from:
 
 It exists so authors can test the workflow-facing contract before composing it.
 
+The workflow-facing MCP surface now has dedicated discovery tools for the
+authoring loop:
+
+- `wf.workflow.list_capabilities`
+  - lists enabled planner-visible workflow-ready node specs
+- `wf.workflow.inspect_capability`
+  - returns one full workflow capability contract with schemas and outcomes
+- `wf.workflow.call_capability`
+  - executes one such capability once for direct testing
+
+These are authoring-plane tools. They do not replace the privileged
+`wf.admin.list_sources` source inventory, and older planner-catalog projections
+may remain while callers migrate to the workflow-facing surface.
+
 ## Relationship To Capability Sources
 
 Sources own capability kinds:

src/wf_mcp/workflow_surface/handlers.py

@@ -44,6 +44,29 @@ class WorkflowSurfaceHandlers:
         ]
         return {"nodes": entries}
 
+    async def list_capabilities(self) -> dict[str, Any]:
+        """Return planner-visible workflow-ready node spec contracts."""
+        capabilities = [
+            detail.model_dump(mode="json")
+            for source in sorted(
+                self.service.capability_sources.values(),
+                key=lambda source: source.id,
+            )
+            if source.enabled and source.visibility.planner
+            for detail in source.as_inventory().capabilities.node_spec_details
+        ]
+        return {"capabilities": capabilities}
+
+    async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
+        """Return one planner-visible workflow capability contract."""
+        for source in self.service.capability_sources.values():
+            if not source.enabled or not source.visibility.planner:
+                continue
+            for detail in source.as_inventory().capabilities.node_spec_details:
+                if detail.name == qualified_name:
+                    return detail.model_dump(mode="json")
+        raise KeyError(f"unknown workflow capability {qualified_name!r}")
+
     async def call_capability(
         self,
         *,

src/wf_mcp/workflow_surface/tools.py

@@ -24,6 +24,22 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
     async def list_artifacts() -> dict[str, Any]:
         return await handlers.list_artifacts()
 
+    @server.tool(
+        name="wf.workflow.list_capabilities",
+        title="List Workflow Capabilities",
+        description="List planner-visible workflow-ready node capabilities.",
+    )
+    async def list_capabilities() -> dict[str, Any]:
+        return await handlers.list_capabilities()
+
+    @server.tool(
+        name="wf.workflow.inspect_capability",
+        title="Inspect Workflow Capability",
+        description="Return one planner-visible workflow capability contract.",
+    )
+    async def inspect_capability(qualified_name: str) -> dict[str, Any]:
+        return await handlers.inspect_capability(qualified_name=qualified_name)
+
     @server.tool(
         name="wf.workflow.call_capability",
         title="Call Workflow Capability",

tests/wf_mcp/test_server.py

@@ -51,6 +51,8 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.admin.call_tool" in names
             assert "wf.admin.get_events" in names
             assert "wf.workflow.list_artifacts" in names
+            assert "wf.workflow.list_capabilities" in names
+            assert "wf.workflow.inspect_capability" in names
             assert "wf.workflow.call_capability" in names
             assert "wf.workflow.run_deployment" in names
 

tests/wf_mcp/test_workflow_surface.py

@@ -74,6 +74,31 @@ def test_workflow_surface_lists_artifact_catalog_entries() -> None:
     assert "plan" not in nodes[0]
 
 
+def test_workflow_surface_lists_planner_visible_capabilities() -> None:
+    handlers = _handlers(FileWorkflowArtifactStore(local_temp_root() / "surface_caps"))
+
+    payload = asyncio.run(handlers.list_capabilities())
+    names = [capability["name"] for capability in payload["capabilities"]]
+
+    assert "wf.std.runtime_error" in names
+    assert "wf.mcp.call_tool" in names
+    assert "wf.admin.list_sources" not in names
+
+
+def test_workflow_surface_inspects_one_capability() -> None:
+    handlers = _handlers(
+        FileWorkflowArtifactStore(local_temp_root() / "surface_inspect_cap")
+    )
+
+    payload = asyncio.run(
+        handlers.inspect_capability(qualified_name="wf.std.runtime_error")
+    )
+
+    assert payload["name"] == "wf.std.runtime_error"
+    assert payload["outcomes"] == ["ok"]
+    assert "input_schema" in payload
+
+
 def test_workflow_surface_validates_deployment_dependencies() -> None:
     artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_validate")
     artifact_store.save_artifact(_artifact())

wf_mcp.config.json

@@ -5,7 +5,7 @@
       "id": "context7.default",
       "server": "context7",
       "account": "default",
-      "enabled": false,
+      "enabled": true,
       "metadata": {
         "transport": "stdio",
         "command": "pnpx",