feat: expose canonical input bindings to mcp

src/wf_mcp/proxy/runtime.py

@@ -48,6 +48,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.validate_draft_workspace",
     "wf.workflow.set_draft_name",
     "wf.workflow.set_draft_route",
+    "wf.workflow.set_step_input_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

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

src/wf_mcp/workflow_surface/tools.py

@@ -35,6 +35,7 @@ from .models import (
     RunDeploymentResult,
     SetDraftNameRequest,
     SetDraftRouteRequest,
+    SetStepInputBindingsRequest,
     SetStepInputMapRequest,
     SetStepOutputMapRequest,
     SetWorkflowOutputMapRequest,
@@ -442,6 +443,27 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             )
         )
 
+    @server.tool(
+        name="wf.workflow.set_step_input_bindings",
+        title="Set Step Input Bindings",
+        description=(
+            "Replace one capability step's complete canonical input-binding list "
+            "atomically. Supports graph-path and literal bindings; inspect the "
+            "draft first because replacement is not a merge."
+        ),
+    )
+    async def set_step_input_bindings(
+        request: SetStepInputBindingsRequest,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.set_step_input_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

@@ -54,6 +54,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.validate_draft_workspace" in names
             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_input_map" in names
             assert "wf.workflow.set_step_output_map" in names
             assert "wf.workflow.set_workflow_output_map" in names
@@ -118,6 +119,12 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             ].inputSchema
             set_input_request = set_input_schema["properties"]["request"]
             assert "merge" in set_input_request["properties"]
+            set_bindings_schema = tools_by_name[
+                "wf.workflow.set_step_input_bindings"
+            ].inputSchema
+            set_bindings_request = set_bindings_schema["properties"]["request"]
+            assert "bindings" in set_bindings_request["properties"]
+            assert "merge" not in set_bindings_request["properties"]
             set_workflow_output_schema = tools_by_name[
                 "wf.workflow.set_workflow_output_map"
             ].inputSchema

tests/wf_mcp/server/test_tools.py

@@ -39,6 +39,7 @@ async def test_server_search_mode_pins_stable_control_and_workflow_tools() -> No
         assert "wf.workflow.validate_draft_workspace" in names
         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_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

@@ -8,13 +8,16 @@ from wf_artifacts import (
     FileWorkflowArtifactStore,
     WorkflowDeployment,
 )
-from wf_core.models.steps import InputPathBinding, OutputBinding
+from wf_core.models.steps import InputPathBinding, InputValueBinding, OutputBinding
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 from wf_mcp.broker import WfMcpService
 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 CreateMinimalDraftWorkspaceRequest
+from wf_mcp.workflow_surface.models import (
+    CreateMinimalDraftWorkspaceRequest,
+    SetStepInputBindingsRequest,
+)
 
 from ..test_support import echo_tool
 from .conftest import (
@@ -269,6 +272,79 @@ def test_minimal_draft_request_accepts_structural_error_message_source() -> None
     assert request.error_message_source.parts == ("error_message",)
 
 
+def test_set_step_input_bindings_request_accepts_path_value_and_null() -> None:
+    request = SetStepInputBindingsRequest.model_validate(
+        {
+            "workspace_id": "concat_draft",
+            "revision": 1,
+            "step_id": "call",
+            "bindings": [
+                {
+                    "target": {"root": "local", "parts": ["items"]},
+                    "path": {"root": "input", "parts": ["items"]},
+                },
+                {
+                    "target": {"root": "local", "parts": ["separator"]},
+                    "value": None,
+                },
+            ],
+        }
+    )
+
+    assert isinstance(request.bindings[0], InputPathBinding)
+    assert request.bindings[0].path == GraphSourcePath.input("items")
+    assert isinstance(request.bindings[1], InputValueBinding)
+    assert request.bindings[1].value is None
+
+
+def test_workflow_surface_sets_ordered_canonical_step_input_bindings(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "surface_input_bindings")
+    service = WfMcpService(
+        store=FileStore(tmp_path / "surface_input_bindings_mcp"),
+        artifact_store=artifact_store,
+        draft_workspace_store=FileDraftWorkspaceStore(
+            tmp_path / "surface_input_bindings_mcp"
+        ),
+    )
+    h = WorkflowSurfaceHandlers(service)
+    created = asyncio.run(
+        h.create_draft_workspace_from_capability(
+            workspace_id="concat_draft",
+            capability_name="wf.std.concat",
+            name="concat_draft",
+        )
+    )
+
+    result = asyncio.run(
+        h.set_step_input_bindings(
+            workspace_id="concat_draft",
+            revision=created["revision"],
+            step_id="call",
+            bindings=[
+                InputPathBinding(
+                    target=LocalPath.of("items"),
+                    path=GraphSourcePath.input("items"),
+                ),
+                InputValueBinding(
+                    target=LocalPath.of("separator"),
+                    value="\n",
+                ),
+            ],
+        )
+    )
+    inspected = asyncio.run(
+        h.get_draft_workspace(workspace_id="concat_draft", include_draft=True)
+    )
+
+    assert result["revision"] == created["revision"] + 1
+    assert inspected["draft"]["steps"]["call"]["input"] == [
+        {"target": "items", "path": "input.items"},
+        {"target": "separator", "value": "\n"},
+    ]
+
+
 def test_workflow_surface_accepts_canonical_bindings_for_minimal_workspace(
     tmp_path: Path,
 ) -> None: