feat: expose canonical output bindings to mcp

src/wf_mcp/proxy/runtime.py

@@ -49,6 +49,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.set_draft_name",
     "wf.workflow.set_draft_route",
     "wf.workflow.set_step_input_bindings",
+    "wf.workflow.set_step_output_bindings",
     "wf.workflow.set_step_input_map",
     "wf.workflow.set_step_output_map",
     "wf.workflow.set_workflow_output_map",

src/wf_mcp/workflow_surface/models.py

@@ -239,6 +239,15 @@ class SetStepInputBindingsRequest(BaseModel):
     bindings: DraftInputBindings
 
 
+class SetStepOutputBindingsRequest(BaseModel):
+    """Replace one step's complete ordered canonical output-binding list."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected current workspace revision.")
+    step_id: NonEmptyString
+    bindings: DraftOutputBindings
+
+
 class SetStepOutputMapRequest(BaseModel):
     """Typed MCP request for replacing or merging one step output map."""
 

src/wf_mcp/workflow_surface/tools.py

@@ -37,6 +37,7 @@ from .models import (
     SetDraftRouteRequest,
     SetStepInputBindingsRequest,
     SetStepInputMapRequest,
+    SetStepOutputBindingsRequest,
     SetStepOutputMapRequest,
     SetWorkflowOutputMapRequest,
     TraceRange,
@@ -464,6 +465,26 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             )
         )
 
+    @server.tool(
+        name="wf.workflow.set_step_output_bindings",
+        title="Set Step Output Bindings",
+        description=(
+            "Replace one capability step's complete ordered output bindings. "
+            "Repeated sources are valid fan-out; state targets must not overlap."
+        ),
+    )
+    async def set_step_output_bindings(
+        request: SetStepOutputBindingsRequest,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.set_step_output_bindings(
+                workspace_id=request.workspace_id,
+                revision=request.revision,
+                step_id=request.step_id,
+                bindings=request.bindings,
+            )
+        )
+
     @server.tool(
         name="wf.workflow.set_step_output_map",
         title="Set Step Output Map",

tests/wf_mcp/server/test_config.py

@@ -55,6 +55,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.set_draft_name" in names
             assert "wf.workflow.set_draft_route" in names
             assert "wf.workflow.set_step_input_bindings" in names
+            assert "wf.workflow.set_step_output_bindings" in names
             assert "wf.workflow.set_step_input_map" in names
             assert "wf.workflow.set_step_output_map" in names
             assert "wf.workflow.set_workflow_output_map" in names
@@ -125,6 +126,14 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             set_bindings_request = set_bindings_schema["properties"]["request"]
             assert "bindings" in set_bindings_request["properties"]
             assert "merge" not in set_bindings_request["properties"]
+            set_output_bindings_schema = tools_by_name[
+                "wf.workflow.set_step_output_bindings"
+            ].inputSchema
+            set_output_bindings_request = set_output_bindings_schema["properties"][
+                "request"
+            ]
+            assert "bindings" in set_output_bindings_request["properties"]
+            assert "merge" not in set_output_bindings_request["properties"]
             set_workflow_output_schema = tools_by_name[
                 "wf.workflow.set_workflow_output_map"
             ].inputSchema

tests/wf_mcp/server/test_tools.py

@@ -40,6 +40,7 @@ async def test_server_search_mode_pins_stable_control_and_workflow_tools() -> No
         assert "wf.workflow.set_draft_name" in names
         assert "wf.workflow.set_draft_route" in names
         assert "wf.workflow.set_step_input_bindings" in names
+        assert "wf.workflow.set_step_output_bindings" in names
         assert "wf.workflow.set_step_input_map" in names
         assert "wf.workflow.set_step_output_map" in names
         assert "wf.workflow.set_workflow_output_map" in names

tests/wf_mcp/workflow_surface/test_drafts.py

@@ -3,6 +3,9 @@ from __future__ import annotations
 import asyncio
 from pathlib import Path
 
+import pytest
+from pydantic import ValidationError
+
 from wf_artifacts import (
     FileDraftWorkspaceStore,
     FileWorkflowArtifactStore,
@@ -17,6 +20,7 @@ from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
 from wf_mcp.workflow_surface.models import (
     CreateMinimalDraftWorkspaceRequest,
     SetStepInputBindingsRequest,
+    SetStepOutputBindingsRequest,
 )
 
 from ..test_support import echo_tool
@@ -297,6 +301,82 @@ def test_set_step_input_bindings_request_accepts_path_value_and_null() -> None:
     assert request.bindings[1].value is None
 
 
+def test_set_step_output_bindings_request_preserves_ordered_source_fan_out() -> None:
+    request = SetStepOutputBindingsRequest.model_validate(
+        {
+            "workspace_id": "draft-output",
+            "revision": 4,
+            "step_id": "analyze",
+            "bindings": [
+                {"source": "report.title", "target": "state.report.title"},
+                {"source": "report.title", "target": "state.audit.title"},
+            ],
+        }
+    )
+
+    assert [str(binding.source) for binding in request.bindings] == [
+        "report.title",
+        "report.title",
+    ]
+
+
+def test_set_step_output_bindings_request_rejects_malformed_canonical_record() -> None:
+    with pytest.raises(ValidationError):
+        SetStepOutputBindingsRequest.model_validate(
+            {
+                "workspace_id": "draft-output",
+                "revision": 4,
+                "step_id": "analyze",
+                "bindings": [{"source": "report.title"}],
+            }
+        )
+
+
+def test_workflow_surface_sets_ordered_canonical_step_output_bindings(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "surface_output_bindings")
+    service = WfMcpService(
+        store=FileStore(tmp_path / "surface_output_bindings_mcp"),
+        artifact_store=artifact_store,
+        draft_workspace_store=FileDraftWorkspaceStore(
+            tmp_path / "surface_output_bindings_mcp"
+        ),
+    )
+    h = WorkflowSurfaceHandlers(service)
+    created = asyncio.run(
+        h.create_minimal_draft_workspace(
+            workspace_id="echo_draft",
+            name="echo_draft",
+            capability_name="wf.std.constant",
+            input_schema={"type": "object"},
+            state_schema={"fields": {"report": {}, "audit": {}}},
+            output_schema={"type": "object"},
+        )
+    )
+
+    result = asyncio.run(
+        h.set_step_output_bindings(
+            workspace_id="echo_draft",
+            revision=created["revision"],
+            step_id="call",
+            bindings=[
+                OutputBinding(source="value", target="state.report"),
+                OutputBinding(source="value", target="state.audit"),
+            ],
+        )
+    )
+    inspected = asyncio.run(
+        h.get_draft_workspace(workspace_id="echo_draft", include_draft=True)
+    )
+
+    assert result["revision"] == created["revision"] + 1
+    assert inspected["draft"]["steps"]["call"]["output"] == [
+        {"source": "value", "target": "state.report"},
+        {"source": "value", "target": "state.audit"},
+    ]
+
+
 def test_workflow_surface_sets_ordered_canonical_step_input_bindings(
     tmp_path: Path,
 ) -> None: