feat: expose capability step updates to mcp

docs/superpowers/plans/2026-07-26-capability-step-update.md

@@ -577,7 +577,7 @@ git commit -m "feat: expose capability step updates over rpc"
 - Produces: `wf.workflow.update_capability_step`.
 - Extends: `wf.workflow.add_step_from_capability`.
 
-- [ ] **Step 1: Write failing MCP request tests**
+- [x] **Step 1: Write failing MCP request tests**
 
 Add:
 
@@ -602,7 +602,7 @@ assert isinstance(request.update.input[0], InputValueBinding)
 
 Add malformed update cases and add-capability request parity tests.
 
-- [ ] **Step 2: Write failing discovery and real invocation tests**
+- [x] **Step 2: Write failing discovery and real invocation tests**
 
 Require the tool in normal and search-mode inventories. Call it through an
 in-memory FastMCP client and assert:
@@ -614,7 +614,7 @@ in-memory FastMCP client and assert:
 
 Extend the add-capability invocation test with metadata and a literal binding.
 
-- [ ] **Step 3: Implement MCP request and tools**
+- [x] **Step 3: Implement MCP request and tools**
 
 Add:
 
@@ -654,7 +654,7 @@ Add the name to `_SEARCH_ALWAYS_VISIBLE_TOOL_NAMES`. Extend
 `AddStepFromCapabilityRequest` and its tool delegation with the creation-parity
 fields.
 
-- [ ] **Step 4: Verify and commit Task 3**
+- [x] **Step 4: Verify and commit Task 3**
 
 Run:
 

src/wf_mcp/proxy/runtime.py

@@ -55,6 +55,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.set_workflow_output_bindings",
     "wf.workflow.set_workflow_output_map",
     "wf.workflow.bind",
+    "wf.workflow.update_capability_step",
     "wf.workflow.remove_draft_route",
     "wf.workflow.remove_draft_step",
     "wf.workflow.remove_draft_binding",

src/wf_mcp/workflow_surface/models.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
-from typing import Annotated, Any, Literal
+from typing import Annotated, Any, Literal, Self
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
+from wf_api import CapabilityStepUpdate
 from wf_api.next_actions import NextActionPatchExample, NextActions
 from wf_artifacts import ArtifactKind
 from wf_artifacts.draft_workspaces.models import WORKSPACE_ID_PATTERN
@@ -342,14 +343,39 @@ class AddStepFromCapabilityRequest(BaseModel):
             "require explicit routes."
         ),
     )
-    input_map: DraftPathMap = Field(
+    input_map: DraftPathMap | None = Field(
-        default_factory=dict,
+        default=None,
         description="Graph source path to rootless node-local target path.",
     )
+    input_bindings: DraftInputBindings | None = Field(
+        default=None,
+        description=(
+            "Preferred complete ordered canonical input bindings. Use this "
+            "instead of input_map for new clients."
+        ),
+    )
     bind_outputs: DraftPathMap = Field(
         default_factory=dict,
         description="Node-local output field to state path with schema projection.",
     )
+    desc: str | None = Field(default=None, min_length=1)
+    retry: int | None = Field(default=None, ge=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
+
+    @model_validator(mode="after")
+    def reject_both_input_forms(self) -> Self:
+        if self.input_map is not None and self.input_bindings is not None:
+            raise ValueError("input_map and input_bindings are mutually exclusive")
+        return self
+
+
+class UpdateCapabilityStepRequest(BaseModel):
+    """Typed MCP request for patching one capability-backed draft step."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected workspace revision.")
+    step_id: NonEmptyString = Field(description="Existing draft step id.")
+    update: CapabilityStepUpdate
 
 
 class RemoveDraftRouteRequest(BaseModel):

src/wf_mcp/workflow_surface/tools.py

@@ -42,6 +42,7 @@ from .models import (
     SetWorkflowOutputBindingsRequest,
     SetWorkflowOutputMapRequest,
     TraceRange,
+    UpdateCapabilityStepRequest,
     ValidateDeploymentResult,
     ValidateDraftWorkspaceRequest,
 )
@@ -591,7 +592,31 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
                 route_from_outcome=request.route_from_outcome,
                 routes=request.routes,
                 input_map=request.input_map,
+                input_bindings=request.input_bindings,
                 bind_outputs=request.bind_outputs,
+                desc=request.desc,
+                retry=request.retry,
+                timeout_seconds=request.timeout_seconds,
+            )
+        )
+
+    @server.tool(
+        name="wf.workflow.update_capability_step",
+        title="Update Capability Step",
+        description=(
+            "Update capability-step metadata and optionally replace its complete "
+            "canonical input bindings. Preserves use, routes, and outputs."
+        ),
+    )
+    async def update_capability_step(
+        request: UpdateCapabilityStepRequest,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.update_capability_step(
+                workspace_id=request.workspace_id,
+                revision=request.revision,
+                step_id=request.step_id,
+                update=request.update,
             )
         )
 

tests/wf_mcp/server/test_config.py

@@ -62,6 +62,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.set_workflow_output_map" in names
             assert "wf.workflow.bind" in names
             assert "wf.workflow.add_step_from_capability" in names
+            assert "wf.workflow.update_capability_step" in names
             assert "wf.workflow.remove_draft_route" in names
             assert "wf.workflow.remove_draft_step" in names
             assert "wf.workflow.remove_draft_binding" in names
@@ -169,6 +170,15 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             add_step_request = add_step_schema["properties"]["request"]
             assert "capability_name" in add_step_request["properties"]
             assert "bind_outputs" in add_step_request["properties"]
+            assert "input_bindings" in add_step_request["properties"]
+            assert "desc" in add_step_request["properties"]
+            assert "retry" in add_step_request["properties"]
+            assert "timeout_seconds" in add_step_request["properties"]
+            update_step_schema = tools_by_name[
+                "wf.workflow.update_capability_step"
+            ].inputSchema
+            update_step_request = update_step_schema["properties"]["request"]
+            assert "update" in update_step_request["properties"]
             from_capability_output = tools_by_name[
                 "wf.workflow.create_draft_workspace_from_capability"
             ].outputSchema

tests/wf_mcp/server/test_tools.py

@@ -55,6 +55,7 @@ async def test_server_search_mode_pins_stable_control_and_workflow_tools() -> No
         assert "wf.workflow.set_workflow_output_bindings" in names
         assert "wf.workflow.set_workflow_output_map" in names
         assert "wf.workflow.bind" in names
+        assert "wf.workflow.update_capability_step" in names
         assert "wf.workflow.remove_draft_route" in names
         assert "wf.workflow.remove_draft_step" in names
         assert "wf.workflow.remove_draft_binding" in names
@@ -208,6 +209,115 @@ async def test_registered_workflow_output_bindings_tool_preserves_union_order(
     ]
 
 
+@pytest.mark.asyncio
+async def test_registered_capability_tools_delegate_presence_aware_requests(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class RecordingWorkflowHandler:
+        def __init__(self) -> None:
+            self.update_calls: list[dict[str, Any]] = []
+            self.add_calls: list[dict[str, Any]] = []
+
+        async def update_capability_step(self, **kwargs: Any) -> dict[str, Any]:
+            self.update_calls.append(kwargs)
+            return {
+                "workspace_id": kwargs["workspace_id"],
+                "revision": kwargs["revision"] + 1,
+                "status": "valid",
+                "diagnostics": [],
+                "summary": {},
+            }
+
+        async def add_step_from_capability(self, **kwargs: Any) -> dict[str, Any]:
+            self.add_calls.append(kwargs)
+            return {
+                "workspace_id": kwargs["workspace_id"],
+                "revision": kwargs["revision"] + 1,
+                "status": "valid",
+                "diagnostics": [],
+                "summary": {},
+            }
+
+    recorder = RecordingWorkflowHandler()
+    monkeypatch.setattr(
+        "wf_mcp.workflow_surface.tools.WorkflowApi",
+        lambda _context: recorder,
+    )
+    service = WfMcpService(
+        store=FileStore(tmp_path / "capability_tool_store"),
+        artifact_store=FileWorkflowArtifactStore(
+            tmp_path / "capability_tool_artifacts"
+        ),
+        draft_workspace_store=FileDraftWorkspaceStore(
+            tmp_path / "capability_tool_drafts"
+        ),
+    )
+    server = FastMCP("capability-update-tool-test")
+    register_workflow_tools(server, service)
+
+    async with Client(FastMCPTransport(server)) as client:
+        updated = await client.call_tool(
+            "wf.workflow.update_capability_step",
+            {
+                "request": {
+                    "workspace_id": "report",
+                    "revision": 4,
+                    "step_id": "publish",
+                    "update": {
+                        "desc": None,
+                        "input": [
+                            {
+                                "value": "markdown",
+                                "target": "request.format",
+                            }
+                        ],
+                    },
+                }
+            },
+        )
+        added = await client.call_tool(
+            "wf.workflow.add_step_from_capability",
+            {
+                "request": {
+                    "workspace_id": "report",
+                    "revision": 5,
+                    "step_id": "archive",
+                    "capability_name": "local.report.archive",
+                    "input_bindings": [
+                        {
+                            "path": "state.report.title",
+                            "target": "request.title",
+                        },
+                        {"value": "pdf", "target": "request.format"},
+                    ],
+                    "desc": "Archive report",
+                    "retry": 0,
+                    "timeout_seconds": 20,
+                }
+            },
+        )
+
+    assert structured(updated)["revision"] == 5
+    assert structured(added)["revision"] == 6
+    assert len(recorder.update_calls) == 1
+    update = recorder.update_calls[0]["update"]
+    assert update.model_fields_set == {"desc", "input"}
+    assert isinstance(update.input[0], InputValueBinding)
+    assert len(recorder.add_calls) == 1
+    add_call = recorder.add_calls[0]
+    assert add_call["input_map"] is None
+    assert [
+        binding.model_dump(mode="json") for binding in add_call["input_bindings"]
+    ] == [
+        {"path": "state.report.title", "target": "request.title"},
+        {"value": "pdf", "target": "request.format"},
+    ]
+    assert add_call["desc"] == "Archive report"
+    assert add_call["retry"] == 0
+    assert add_call["timeout_seconds"] == 20
+
+
 @pytest.mark.asyncio
 async def test_server_search_mode_can_use_safe_tool_names() -> None:
     config = server_config()

tests/wf_mcp/workflow_surface/test_drafts.py

@@ -18,10 +18,12 @@ from wf_mcp.models import ConnectionConfig
 from wf_mcp.storage import FileStore
 from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
 from wf_mcp.workflow_surface.models import (
+    AddStepFromCapabilityRequest,
     CreateMinimalDraftWorkspaceRequest,
     SetStepInputBindingsRequest,
     SetStepOutputBindingsRequest,
     SetWorkflowOutputBindingsRequest,
+    UpdateCapabilityStepRequest,
 )
 
 from ..test_support import echo_tool
@@ -33,6 +35,91 @@ from .conftest import (
 )
 
 
+def test_update_capability_step_request_preserves_field_presence_and_binding_type() -> (
+    None
+):
+    request = UpdateCapabilityStepRequest.model_validate(
+        {
+            "workspace_id": "report",
+            "revision": 4,
+            "step_id": "publish",
+            "update": {
+                "desc": None,
+                "input": [
+                    {"value": "markdown", "target": "request.format"},
+                ],
+            },
+        }
+    )
+
+    assert request.update.model_fields_set == {"desc", "input"}
+    assert request.update.desc is None
+    assert request.update.input is not None
+    assert isinstance(request.update.input[0], InputValueBinding)
+
+
+@pytest.mark.parametrize(
+    "update",
+    [
+        {},
+        {"input": None},
+        {"retry": -1},
+        {"timeout_seconds": 0},
+        {"unknown": "field"},
+    ],
+)
+def test_update_capability_step_request_rejects_invalid_patch(update) -> None:
+    with pytest.raises(ValidationError):
+        UpdateCapabilityStepRequest.model_validate(
+            {
+                "workspace_id": "report",
+                "revision": 4,
+                "step_id": "publish",
+                "update": update,
+            }
+        )
+
+
+def test_add_step_from_capability_request_preserves_creation_parity() -> None:
+    request = AddStepFromCapabilityRequest.model_validate(
+        {
+            "workspace_id": "report",
+            "revision": 3,
+            "step_id": "publish",
+            "capability_name": "local.report.publish",
+            "input_bindings": [
+                {"path": "state.report.title", "target": "request.title"},
+                {"value": "markdown", "target": "request.format"},
+            ],
+            "desc": "Publish report",
+            "retry": 0,
+            "timeout_seconds": 30,
+        }
+    )
+
+    assert request.input_map is None
+    assert request.input_bindings is not None
+    assert isinstance(request.input_bindings[0], InputPathBinding)
+    assert isinstance(request.input_bindings[1], InputValueBinding)
+    assert request.retry == 0
+
+
+def test_add_step_from_capability_request_rejects_both_input_forms() -> None:
+    with pytest.raises(ValidationError, match="mutually exclusive"):
+        AddStepFromCapabilityRequest.model_validate(
+            {
+                "workspace_id": "report",
+                "revision": 3,
+                "step_id": "publish",
+                "capability_name": "local.report.publish",
+                "input_map": {"state.title": "request.title"},
+                "input_bindings": [
+                    {"value": "markdown", "target": "request.format"},
+                ],
+            }
+        )
+
+
 def test_workflow_surface_rejects_unknown_draft_route_outcome_when_spec_is_known(
     tmp_path: Path,
 ) -> None: