feat: add draft step capability helper

docs/current_roadmap.md

@@ -69,6 +69,10 @@ clear operator feedback before adding more architecture.
 - Completed: `wf draft bind-output-to-state` composes state schema projection
   with output binding merge, reducing manual draft patch repairs in agent
   challenge runs.
+- Completed: `wf draft add-step-from-capability` inserts one explicit
+  capability-backed step with route, input, and output-to-state schema/binding
+  wiring in a single revision, reducing brittle JSON Patch authoring for
+  multi-step workflows.
 - Completed: draft validation now preserves structured core validation issues
   and adds exact `bind-output-to-state` repair hints for missing state fields.
 - Keep status read-only; do not mutate registry, auth, config, or stores.

docs/wf_cli.md

@@ -339,6 +339,27 @@ The command combines two common edits:
 
 Use `set-route` separately for outcome routing.
 
+### Add A Capability Step To A Draft
+
+Use `wf draft add-step-from-capability` when adding a new capability-backed step
+to an existing draft. The command is explicit: it does not guess missing maps.
+
+```bash
+wf draft add-step-from-capability report_ws \
+  --revision 3 \
+  --step render \
+  --capability local.report.render_markdown_report \
+  --from-step extract \
+  --from-outcome ok \
+  --outcome ok \
+  --to __end__ \
+  --input state.title=title \
+  --bind-output markdown=state.markdown
+```
+
+Run `wf draft validate report_ws` after adding the step. If validation returns
+a `repair_hint`, prefer the focused helper in that hint before JSON Patch.
+
 Validate:
 
 ```bash

skills/wf-cli/SKILL.md

@@ -46,6 +46,7 @@ wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=st
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
 wf draft add-state-from-output <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
 wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
+wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --outcome ok --to <next-or-__end__> --input input.text=text --bind-output result=state.result
 wf draft validate <workspace_id>
 wf draft save <workspace_id> --artifact <artifact_id> --version <n> --title <title>
 
@@ -90,6 +91,10 @@ when you need the schema declaration without changing bindings.
 `bind-output-to-state` requires a capability-backed step with `use`; use JSON
 Patch for non-capability/control draft steps.
 
+To add a capability step, prefer `wf draft add-step-from-capability` over raw
+JSON Patch when the route, input bindings, and output-to-state bindings are
+known. It is explicit and does not guess missing maps.
+
 ## Rules
 
 - Use explicit `--config <path>` for examples, challenge workspaces, and

skills/wf-workflow/references/draft-workspaces.md

@@ -75,6 +75,7 @@ Prefer focused helpers over JSON Patch for common edits:
 - `set_step_output_map`
 - `add_state_schema_from_output`
 - `bind_output_to_state`
+- `add_step_from_capability`
 
 CLI equivalents:
 
@@ -87,6 +88,7 @@ wf draft set-output <workspace_id> --revision <n> --step <step_id> --map text=st
 wf draft set-output <workspace_id> --revision <n> --step <step_id> --merge --map other=state.other
 wf draft add-state-from-output <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
 wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --output <field> --state state.<field>
+wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --outcome ok --to <next-or-__end__> --input input.text=text --bind-output result=state.result
 ```
 
 `set-input` direction: `input.text=text` means graph source `input.text` maps to
@@ -118,6 +120,19 @@ wf draft bind-output-to-state <workspace_id> --revision <n> --step <step_id> --o
 wf draft validate <workspace_id>
 ```
 
+- `add_step_from_capability`
+
+  Adds a new capability-backed step with explicit route, input bindings, and
+  output-to-state schema/binding wiring in one revision. It can set the incoming
+  edge, outgoing edge, input map, and output-to-state schema/binding. It still
+  requires explicit choices; if you do not know a map, inspect the capability or
+  run validation rather than guessing.
+
+```bash
+wf draft add-step-from-capability <workspace_id> --revision <n> --step <step_id> --capability <qualified_name> --from-step <prev> --from-outcome ok --outcome ok --to <next-or-__end__> --input input.text=text --bind-output result=state.result
+wf draft validate <workspace_id>
+```
+
 Validation repair hints are product guidance. If a diagnostic suggests
 `bind-output-to-state`, use it before hand-editing `state_schema` or step output
 bindings.

skills/wf-workflow/references/workflow-lifecycle.md

@@ -23,6 +23,13 @@ validated, runnable deployment.
    - Use `bind-output-to-state` when a capability output should become state;
      it declares the matching state schema and merges the output binding in one
      revision-checked edit.
+   - When adding a new capability-backed step, prefer:
+     ```bash
+     wf draft add-step-from-capability ...
+     wf draft validate <workspace_id>
+     ```
+     Use raw `wf draft patch` only when changing structure that no focused helper
+     covers.
    - Use JSON Patch only for general structural edits.
 6. Save an artifact.
    - Draft artifact:

src/wf_api/drafts.py

@@ -389,6 +389,96 @@ class WorkflowDraftApi:
             ],
         )
 
+    async def add_step_from_capability(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        capability_name: str,
+        route_from_step: str | None = None,
+        route_from_outcome: str = DEFAULT_OK_OUTCOME,
+        route_outcome: str = DEFAULT_OK_OUTCOME,
+        route_to: str = "__end__",
+        input_map: dict[str, str] | None = None,
+        bind_outputs: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        """Add one capability step plus explicit route/map/schema wiring.
+
+        This is a composed authoring helper for agents.  It edits the draft in
+        one revision so callers do not have to interleave add-step, route,
+        input-map, state-schema, and output-map operations by hand.
+        """
+        workspace = self._draft_store().get_workspace(workspace_id)
+        steps = workspace.draft.get("steps")
+        if not isinstance(steps, dict):
+            raise ValueError("draft steps must be an object")
+        if step_id in steps:
+            raise ValueError(f"draft step {step_id!r} already exists")
+
+        spec = self.context.specs.get_qualified_spec(capability_name)
+        output_schema = (
+            spec.output_schema_contract or spec.output_model.model_json_schema()
+        )
+        state_schema = workspace.draft.get("state_schema", {})
+        if not isinstance(state_schema, dict):
+            raise ValueError("draft state_schema must be an object")
+
+        input_map = input_map or {}
+        bind_outputs = bind_outputs or {}
+        projected_state_schema = state_schema
+        for output_field, state_path in bind_outputs.items():
+            state_field = _state_root_field(state_path)
+            projected_state_schema = project_output_property_to_state_schema(
+                state_schema=projected_state_schema,
+                output_schema=output_schema,
+                output_field=output_field,
+                state_field=state_field,
+            )
+
+        patch: list[dict[str, Any]] = [
+            {
+                "op": "add",
+                "path": f"/steps/{_escape_json_pointer(step_id)}",
+                "value": {
+                    "use": capability_name,
+                    "input": _draft_input_bindings_payload(input_map, {}),
+                    "output": _draft_output_bindings_payload(bind_outputs),
+                },
+            },
+            {
+                "op": "add",
+                "path": f"/routes/{_escape_json_pointer(step_id)}",
+                "value": {route_outcome: route_to},
+            },
+        ]
+        if projected_state_schema != state_schema:
+            patch.insert(
+                0,
+                {
+                    "op": "replace",
+                    "path": "/state_schema",
+                    "value": projected_state_schema,
+                },
+            )
+        if route_from_step is not None:
+            patch.append(
+                {
+                    "op": "add",
+                    "path": (
+                        f"/routes/{_escape_json_pointer(route_from_step)}/"
+                        f"{_escape_json_pointer(route_from_outcome)}"
+                    ),
+                    "value": step_id,
+                }
+            )
+
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
     def _step_input_maps(
         self,
         *,

src/wf_api/service.py

@@ -396,6 +396,33 @@ class WorkflowApi:
             state_path=state_path,
         )
 
+    async def add_step_from_capability(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        capability_name: str,
+        route_from_step: str | None = None,
+        route_from_outcome: str = "ok",
+        route_outcome: str = "ok",
+        route_to: str = "__end__",
+        input_map: dict[str, str] | None = None,
+        bind_outputs: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        return await self.drafts.add_step_from_capability(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+            capability_name=capability_name,
+            route_from_step=route_from_step,
+            route_from_outcome=route_from_outcome,
+            route_outcome=route_outcome,
+            route_to=route_to,
+            input_map=input_map,
+            bind_outputs=bind_outputs,
+        )
+
     async def create_minimal_draft_workspace(
         self,
         *,

src/wf_api/surface.py

@@ -134,6 +134,21 @@ class WorkflowDraftSurface(Protocol):
         state_path: str,
     ) -> dict[str, Any]: ...
 
+    async def add_step_from_capability(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        capability_name: str,
+        route_from_step: str | None = None,
+        route_from_outcome: str = "ok",
+        route_outcome: str = "ok",
+        route_to: str = "__end__",
+        input_map: dict[str, str] | None = None,
+        bind_outputs: dict[str, str] | None = None,
+    ) -> dict[str, Any]: ...
+
     async def validate_draft_workspace(
         self,
         *,

src/wf_cli/commands/drafts.py

@@ -359,6 +359,81 @@ def bind_output_to_state(
     )
 
 
+@app.command("add-step-from-capability")
+def add_step_from_capability(
+    ctx: typer.Context,
+    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
+    revision: Annotated[
+        int, typer.Option("--revision", min=1, help="Expected workspace revision.")
+    ],
+    step_id: Annotated[str, typer.Option("--step", help="New draft step id.")],
+    capability_name: Annotated[
+        str, typer.Option("--capability", help="Qualified capability name.")
+    ],
+    route_from_step: Annotated[
+        str | None,
+        typer.Option(
+            "--from-step",
+            help="Optional existing step whose outcome should route to this step.",
+        ),
+    ] = None,
+    route_from_outcome: Annotated[
+        str,
+        typer.Option("--from-outcome", help="Outcome on --from-step."),
+    ] = "ok",
+    route_outcome: Annotated[
+        str,
+        typer.Option("--outcome", help="Outcome emitted by the new step."),
+    ] = "ok",
+    route_to: Annotated[
+        str,
+        typer.Option("--to", help="Target step id or __end__ for the new step."),
+    ] = "__end__",
+    input_mapping: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--input",
+            help="Input binding SOURCE=LOCAL_TARGET. Repeat for multiple inputs.",
+        ),
+    ] = None,
+    output_mapping: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--bind-output",
+            help=(
+                "Output binding LOCAL_OUTPUT=STATE_TARGET with state schema "
+                "projection. Repeat for multiple outputs."
+            ),
+        ),
+    ] = None,
+) -> None:
+    """Add one capability step with explicit route, input, and output wiring.
+
+    This command does not guess missing maps. Pass the route and bindings you
+    want, then run `wf draft validate <workspace_id>`.
+    """
+    input_map = _parse_map_flags(input_mapping)
+    bind_outputs = _parse_map_flags(output_mapping)
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.add_step_from_capability(
+                workspace_id=workspace_id,
+                revision=revision,
+                step_id=step_id,
+                capability_name=capability_name,
+                route_from_step=route_from_step,
+                route_from_outcome=route_from_outcome,
+                route_outcome=route_outcome,
+                route_to=route_to,
+                input_map=input_map,
+                bind_outputs=bind_outputs,
+            ),
+        )
+    )
+
+
 @app.command("validate")
 def validate_draft(
     ctx: typer.Context,

src/wf_mcp/workflow_surface/models.py

@@ -265,6 +265,39 @@ class BindOutputToStateRequest(BaseModel):
     )
 
 
+class AddStepFromCapabilityRequest(BaseModel):
+    """Typed MCP request for adding one capability-backed draft step with wiring."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected workspace revision.")
+    step_id: str = Field(description="New draft step id.")
+    capability_name: str = Field(description="Qualified capability name.")
+    route_from_step: str | None = Field(
+        default=None,
+        description="Optional existing step whose outcome should route to the new step.",
+    )
+    route_from_outcome: str = Field(
+        default="ok",
+        description="Outcome on route_from_step that should route to the new step.",
+    )
+    route_outcome: str = Field(
+        default="ok",
+        description="Outcome emitted by the new step.",
+    )
+    route_to: str = Field(
+        default="__end__",
+        description="Target step id or __end__ for the new step outcome.",
+    )
+    input_map: dict[str, str] = Field(
+        default_factory=dict,
+        description="Graph source path to node-local target field.",
+    )
+    bind_outputs: dict[str, str] = Field(
+        default_factory=dict,
+        description="Node-local output field to state path with schema projection.",
+    )
+
+
 class DeleteDraftWorkspaceRequest(BaseModel):
     """Typed MCP request payload for deleting one draft workspace."""
 

src/wf_mcp/workflow_surface/tools.py

@@ -13,6 +13,7 @@ from wf_mcp.broker.service.workflow_operation_context import context_from_servic
 
 from .models import (
     AddStateFromOutputRequest,
+    AddStepFromCapabilityRequest,
     BindOutputToStateRequest,
     CallCapabilityResult,
     CreateArtifactFromWorkspaceRequest,
@@ -483,6 +484,32 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             )
         )
 
+    @server.tool(
+        name="wf.workflow.add_step_from_capability",
+        title="Add Step From Capability",
+        description=(
+            "Add one capability-backed draft step with explicit route, input, "
+            "and output-to-state binding hints."
+        ),
+    )
+    async def add_step_from_capability(
+        request: AddStepFromCapabilityRequest,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.add_step_from_capability(
+                workspace_id=request.workspace_id,
+                revision=request.revision,
+                step_id=request.step_id,
+                capability_name=request.capability_name,
+                route_from_step=request.route_from_step,
+                route_from_outcome=request.route_from_outcome,
+                route_outcome=request.route_outcome,
+                route_to=request.route_to,
+                input_map=request.input_map,
+                bind_outputs=request.bind_outputs,
+            )
+        )
+
     @server.tool(
         name="wf.workflow.create_minimal_draft_workspace",
         title="Create Minimal Draft Workspace",

src/wf_transport_rpc_http/__init__.py

@@ -4,6 +4,7 @@ from .app import create_rpc_app
 from .client import RpcWorkflowApiClient
 from .errors import WorkflowRpcError
 from .models import (
+    AddStepFromCapabilityParams,
     AdminEmptyParams,
     CallCapabilityParams,
     CreateArtifactFromPlanParams,
@@ -43,6 +44,7 @@ from .models import (
 __all__ = [
     "CreateArtifactFromPlanParams",
     "CreateArtifactFromWorkspaceParams",
+    "AddStepFromCapabilityParams",
     "AdminEmptyParams",
     "CallCapabilityParams",
     "CreateDraftFromCapabilityParams",

src/wf_transport_rpc_http/client/drafts.py

@@ -181,6 +181,36 @@ class RpcDraftClientMixin:
             },
         )
 
+    async def add_step_from_capability(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+        capability_name: str,
+        route_from_step: str | None = None,
+        route_from_outcome: str = "ok",
+        route_outcome: str = "ok",
+        route_to: str = "__end__",
+        input_map: dict[str, str] | None = None,
+        bind_outputs: dict[str, str] | None = None,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.draft_workspaces.add_step_from_capability",
+            {
+                "workspace_id": workspace_id,
+                "revision": revision,
+                "step_id": step_id,
+                "capability_name": capability_name,
+                "route_from_step": route_from_step,
+                "route_from_outcome": route_from_outcome,
+                "route_outcome": route_outcome,
+                "route_to": route_to,
+                "input_map": input_map or {},
+                "bind_outputs": bind_outputs or {},
+            },
+        )
+
     async def validate_draft_workspace(
         self: RpcCaller,
         *,

src/wf_transport_rpc_http/methods/drafts.py

@@ -9,6 +9,7 @@ from wf_server import WorkflowServer
 from ..errors import WorkflowRpcError, raise_workflow_rpc_error
 from ..models import (
     AddStateFromOutputParams,
+    AddStepFromCapabilityParams,
     BindOutputToStateParams,
     CreateArtifactFromWorkspaceParams,
     CreateDraftFromCapabilityParams,
@@ -215,6 +216,29 @@ def register_methods(
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(
+        name="workflow.draft_workspaces.add_step_from_capability",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_draft_workspaces_add_step_from_capability(
+        params: AddStepFromCapabilityParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.add_step_from_capability(
+                workspace_id=params.workspace_id,
+                revision=params.revision,
+                step_id=params.step_id,
+                capability_name=params.capability_name,
+                route_from_step=params.route_from_step,
+                route_from_outcome=params.route_from_outcome,
+                route_outcome=params.route_outcome,
+                route_to=params.route_to,
+                input_map=params.input_map,
+                bind_outputs=params.bind_outputs,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     @entrypoint.method(
         name="workflow.draft_workspaces.validate", errors=[WorkflowRpcError]
     )

src/wf_transport_rpc_http/models.py

@@ -157,6 +157,19 @@ class BindOutputToStateParams(RpcParamsModel):
     state_path: str = Field(min_length=1)
 
 
+class AddStepFromCapabilityParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    step_id: str = Field(min_length=1)
+    capability_name: str = Field(min_length=1)
+    route_from_step: str | None = None
+    route_from_outcome: str = Field(default="ok", min_length=1)
+    route_outcome: str = Field(default="ok", min_length=1)
+    route_to: str = Field(default="__end__", min_length=1)
+    input_map: dict[str, str] = Field(default_factory=dict)
+    bind_outputs: dict[str, str] = Field(default_factory=dict)
+
+
 class ValidateDraftWorkspaceParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
 

tests/wf_api/test_drafts_service.py

@@ -762,3 +762,74 @@ async def test_bind_output_to_state_rejects_step_without_capability_use(
             output_field="after",
             state_path="state.after",
         )
+
+
+@pytest.mark.asyncio
+async def test_add_step_from_capability_wires_route_inputs_and_state_outputs(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_add_step")
+    api, service = _draft_api(artifact_store, register_echo=True)
+    service.register_specs("demo.personal", echo_tool, _snapshot_tool)
+    await api.create_draft_workspace(
+        workspace_id="echo_ws",
+        draft=_echo_draft(),
+    )
+
+    result = await api.add_step_from_capability(
+        workspace_id="echo_ws",
+        revision=1,
+        step_id="snap",
+        capability_name="demo.personal.snapshot_tool",
+        route_from_step="echo",
+        route_from_outcome="ok",
+        route_outcome="ok",
+        route_to="__end__",
+        input_map={},
+        bind_outputs={"after": "state.after"},
+    )
+
+    assert result["revision"] == 2
+    assert result["status"] == "valid"
+    fetched = await api.get_draft_workspace(workspace_id="echo_ws", include_draft=True)
+    draft = fetched["draft"]
+    assert draft["steps"]["snap"]["use"] == "demo.personal.snapshot_tool"
+    assert draft["routes"]["echo"]["ok"] == "snap"
+    assert draft["routes"]["snap"]["ok"] == "__end__"
+    assert draft["steps"]["snap"]["output"] == [
+        {
+            "source": {"root": "local", "parts": ["after"]},
+            "target": {"root": "state", "parts": ["after"]},
+        }
+    ]
+    assert draft["state_schema"]["properties"]["after"]["$ref"] == "#/$defs/_Snapshot"
+    assert draft["state_schema"]["$defs"]["_Snapshot"]["properties"]["clicked"] == {
+        "title": "Clicked",
+        "type": "boolean",
+    }
+
+
+@pytest.mark.asyncio
+async def test_add_step_from_capability_rejects_existing_step_id(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_add_step_duplicate")
+    api, _service = _draft_api(artifact_store, register_echo=True)
+    await api.create_draft_workspace(
+        workspace_id="echo_ws",
+        draft=_echo_draft(),
+    )
+
+    with pytest.raises(ValueError, match="draft step 'echo' already exists"):
+        await api.add_step_from_capability(
+            workspace_id="echo_ws",
+            revision=1,
+            step_id="echo",
+            capability_name="demo.personal.echo_tool",
+            route_from_step=None,
+            route_from_outcome="ok",
+            route_outcome="ok",
+            route_to="__end__",
+            input_map={},
+            bind_outputs={},
+        )

tests/wf_cli/test_app.py

@@ -161,3 +161,12 @@ def test_wf_draft_bind_output_to_state_help_explains_composed_edit() -> None:
     assert "state schema" in result.output
     assert "output binding" in result.output
     assert "validate" in result.output
+
+
+def test_wf_draft_add_step_from_capability_help_explains_explicit_wiring() -> None:
+    result = runner.invoke(app, ["draft", "add-step-from-capability", "--help"])
+
+    assert result.exit_code == 0
+    assert "--from-step" in result.output
+    assert "--bind-output" in result.output
+    assert "does not guess" in result.output

tests/wf_cli/test_remote_target.py

@@ -1203,6 +1203,64 @@ def test_wf_draft_bind_output_to_state_uses_rpc_target(monkeypatch, tmp_path) ->
     assert payload["revision"] == 2
 
 
+def test_wf_draft_add_step_from_capability_uses_rpc_target(
+    monkeypatch, tmp_path
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    _patch_rpc_client_to_server(monkeypatch, server)
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    runner = CliRunner()
+    base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
+
+    created = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "create-from-capability",
+            "add_step_ws",
+            "wf.std.constant",
+            "--name",
+            "add_step",
+        ],
+    )
+    assert created.exit_code == 0, created.output
+
+    result = runner.invoke(
+        app,
+        [
+            *base_args,
+            "draft",
+            "add-step-from-capability",
+            "add_step_ws",
+            "--revision",
+            "1",
+            "--step",
+            "second",
+            "--capability",
+            "wf.std.constant",
+            "--from-step",
+            "call",
+            "--from-outcome",
+            "ok",
+            "--outcome",
+            "ok",
+            "--to",
+            "__end__",
+            "--input",
+            "input.value=value",
+            "--bind-output",
+            "value=state.second_value",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["revision"] == 2
+    assert payload["status"] == "valid"
+
+
 def test_wf_deploy_create_alias_saves_deployment(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     asyncio.run(

tests/wf_mcp/server/test_config.py

@@ -58,6 +58,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.set_step_output_map" in names
             assert "wf.workflow.add_state_from_output" in names
             assert "wf.workflow.bind_output_to_state" in names
+            assert "wf.workflow.add_step_from_capability" in names
             assert "wf.workflow.create_minimal_draft_workspace" in names
             assert "wf.workflow.create_draft_workspace_from_capability" in names
             assert "wf.workflow.create_artifact_from_workspace" in names
@@ -120,6 +121,12 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             state_request = state_from_output_schema["properties"]["request"]
             assert "output_field" in state_request["properties"]
             assert "state_path" in state_request["properties"]
+            add_step_schema = tools_by_name[
+                "wf.workflow.add_step_from_capability"
+            ].inputSchema
+            add_step_request = add_step_schema["properties"]["request"]
+            assert "capability_name" in add_step_request["properties"]
+            assert "bind_outputs" in add_step_request["properties"]
             from_capability_output = tools_by_name[
                 "wf.workflow.create_draft_workspace_from_capability"
             ].outputSchema

tests/wf_transport_rpc_http/test_app.py

@@ -787,6 +787,43 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
     )
 
 
+async def test_rpc_draft_workspace_add_step_from_capability(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        await _rpc(
+            client,
+            "workflow.draft_workspaces.create_from_capability",
+            {
+                "workspace_id": "add_step_ws",
+                "capability_name": "wf.std.constant",
+                "name": "add_step",
+            },
+        )
+
+        response = await _rpc(
+            client,
+            "workflow.draft_workspaces.add_step_from_capability",
+            {
+                "workspace_id": "add_step_ws",
+                "revision": 1,
+                "step_id": "second",
+                "capability_name": "wf.std.constant",
+                "route_from_step": "call",
+                "route_from_outcome": "ok",
+                "route_outcome": "ok",
+                "route_to": "__end__",
+                "input_map": {"input.value": "value"},
+                "bind_outputs": {"value": "state.second_value"},
+            },
+        )
+
+    result = response["result"]
+    assert result["revision"] == 2
+    assert result["status"] == "valid"
+
+
 async def test_rpc_diagnoses_source(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     app = create_rpc_app(server)

tests/wf_transport_rpc_http/test_client.py

@@ -526,6 +526,40 @@ async def test_rpc_client_draft_workspace_focused_edit_methods(tmp_path) -> None
     assert state_bound["revision"] == 9
 
 
+async def test_rpc_client_draft_workspace_add_step_from_capability(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(
+        transport=transport, base_url="http://test"
+    ) as http_client:
+        client = RpcWorkflowApiClient(
+            url="http://test/rpc",
+            timeout_seconds=5,
+            http_client=http_client,
+        )
+        await client.create_draft_workspace_from_capability(
+            workspace_id="client_add_step_ws",
+            capability_name="wf.std.constant",
+            name="client_add_step",
+        )
+        result = await client.add_step_from_capability(
+            workspace_id="client_add_step_ws",
+            revision=1,
+            step_id="second",
+            capability_name="wf.std.constant",
+            route_from_step="call",
+            route_from_outcome="ok",
+            route_outcome="ok",
+            route_to="__end__",
+            input_map={"input.value": "value"},
+            bind_outputs={"value": "state.second_value"},
+        )
+
+    assert result["revision"] == 2
+    assert result["status"] == "valid"
+
+
 async def test_rpc_client_diagnoses_source(tmp_path) -> None:
     calls: list[tuple[str, dict[str, object]]] = []