add call_capability

docs/wf_mcp_proxy_reality_and_roadmap.md

@@ -126,8 +126,8 @@ When the exposed tool catalog grows large or changes often, FastMCP's search
 transform is a good mitigation: keep a stable pinned control/workflow spine
 visible, and use `search_tools` plus its synthetic `call_tool` for the changing
 rest of the catalog. Do not confuse that synthetic raw-tool caller with a
-future workflow-capability test tool; testing a normalized `NodeSpec` contract
+workflow-capability test tool; `wf.workflow.call_capability` tests a normalized
-is a separate operation and should also remain pinned once it exists.
+`NodeSpec` contract and remains pinned for that reason.
 
 ## What Is Probably Not Worth Owning Yet
 

docs/workflow_capabilities.md

@@ -266,11 +266,11 @@ Today:
 - `wf.mcp` owns workflow-facing MCP runtime helpers
 - discovered upstream tools can already become workflow node specs
 - saved artifacts can be tagged with `kind="workflow"` or `kind="wrapper"`
+- `wf.workflow.call_capability` can execute one planner-visible workflow
+  capability directly and return normalized `outcome` / `output`
 
 Not yet implemented:
 
-- a direct public tool for calling arbitrary workflow node specs for authoring
-  tests
 - per-outcome output schemas
 - graph-as-node for saved workflows
 

src/wf_mcp/transparent_proxy/runtime.py

@@ -35,6 +35,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     # Stable workflow control surface. Keep future workflow-capability test
     # tools pinned here too; they are distinct from raw MCP tool execution.
     "wf.workflow.list_artifacts",
+    "wf.workflow.call_capability",
     "wf.workflow.inspect_artifact",
     "wf.workflow.list_deployments",
     "wf.workflow.validate_deployment",

src/wf_mcp/workflow_surface/handlers.py

@@ -14,6 +14,8 @@ from wf_artifacts import (
     create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
     validate_deployment_dependencies,
 )
+from wf_authoring import build_async_registry
+from wf_core import RuntimeContext
 
 from ..events import make_event
 from ..models import RawWorkflowPlan
@@ -39,6 +41,25 @@ class WorkflowSurfaceHandlers:
         ]
         return {"nodes": entries}
 
+    async def call_capability(
+        self,
+        *,
+        qualified_name: str,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Execute one planner-visible workflow capability for authoring tests."""
+        spec = self.service._get_qualified_spec(qualified_name)
+        handler = build_async_registry(spec)[spec.name]
+        result = await handler(
+            payload,
+            RuntimeContext(current_node_id=spec.name),
+        )
+        return {
+            "qualified_name": spec.name,
+            "outcome": result["outcome"],
+            "output": result["output"],
+        }
+
     async def save_artifact(self, artifact: dict[str, Any]) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")

src/wf_mcp/workflow_surface/tools.py

@@ -21,6 +21,23 @@ 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.call_capability",
+        title="Call Workflow Capability",
+        description=(
+            "Execute one planner-visible workflow capability once and return its "
+            "normalized outcome and output."
+        ),
+    )
+    async def call_capability(
+        qualified_name: str,
+        payload: dict[str, Any],
+    ) -> dict[str, Any]:
+        return await handlers.call_capability(
+            qualified_name=qualified_name,
+            payload=payload,
+        )
+
     @server.tool(
         name="wf.workflow.save_artifact",
         title="Save Workflow Artifact",

tests/wf_mcp/test_server.py

@@ -51,6 +51,7 @@ 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.call_capability" in names
             assert "wf.workflow.run_deployment" in names
 
             echo_result = await client.call_tool(
@@ -58,10 +59,19 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
                 {"text": "hello"},
             )
             artifacts_result = await client.call_tool("wf.workflow.list_artifacts")
+            capability_result = await client.call_tool(
+                "wf.workflow.call_capability",
+                {
+                    "qualified_name": "wf.std.constant",
+                    "payload": {"value": "hello"},
+                },
+            )
             sources_result = await client.call_tool("wf.admin.list_sources")
 
             assert _structured(echo_result)["echoed"] == "hello"
             assert _structured(artifacts_result)["nodes"] == []
+            assert _structured(capability_result)["outcome"] == "ok"
+            assert _structured(capability_result)["output"] == {"value": "hello"}
             source_ids = {
                 source["id"] for source in _structured(sources_result)["result"]
             }
@@ -133,6 +143,7 @@ def test_server_search_mode_pins_stable_control_and_workflow_tools() -> None:
             assert "wf.admin.list_proxy_tools" in names
             assert "wf.admin.get_proxy_tool" in names
             assert "wf.workflow.list_artifacts" in names
+            assert "wf.workflow.call_capability" in names
             assert "wf.workflow.inspect_artifact" in names
             assert "wf.workflow.list_deployments" in names
             assert "wf.workflow.validate_deployment" in names