feat: expose workflow output bindings to mcp

src/wf_mcp/proxy/runtime.py

@@ -52,6 +52,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.set_step_output_bindings",
     "wf.workflow.set_step_input_map",
     "wf.workflow.set_step_output_map",
+    "wf.workflow.set_workflow_output_bindings",
     "wf.workflow.set_workflow_output_map",
     "wf.workflow.bind",
     "wf.workflow.remove_draft_route",

src/wf_mcp/workflow_surface/models.py

@@ -57,6 +57,15 @@ DraftOutputBindings = Annotated[
         )
     ),
 ]
+WorkflowOutputBindings = Annotated[
+    list[InputBinding],
+    Field(
+        description=(
+            "Complete ordered public workflow-output projection. Path bindings "
+            "read from input, state, or context; value bindings emit JSON literals."
+        )
+    ),
+]
 JsonPatchOperations = Annotated[
     list[dict[str, Any]],
     Field(description="RFC 6902 JSON Patch operations."),
@@ -281,6 +290,14 @@ class SetWorkflowOutputMapRequest(BaseModel):
     )
 
 
+class SetWorkflowOutputBindingsRequest(BaseModel):
+    """Replace the complete canonical workflow-output binding list."""
+
+    workspace_id: WorkspaceId
+    revision: int = Field(ge=1, description="Expected current workspace revision.")
+    bindings: WorkflowOutputBindings
+
+
 class BindDraftRequest(BaseModel):
     """Typed MCP request for binding one draft step path with schema projection."""
 

src/wf_mcp/workflow_surface/tools.py

@@ -39,6 +39,7 @@ from .models import (
     SetStepInputMapRequest,
     SetStepOutputBindingsRequest,
     SetStepOutputMapRequest,
+    SetWorkflowOutputBindingsRequest,
     SetWorkflowOutputMapRequest,
     TraceRange,
     ValidateDeploymentResult,
@@ -507,6 +508,25 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             )
         )
 
+    @server.tool(
+        name="wf.workflow.set_workflow_output_bindings",
+        title="Set Workflow Output Bindings",
+        description=(
+            "Replace the complete ordered workflow-output binding list with "
+            "canonical path and literal records."
+        ),
+    )
+    async def set_workflow_output_bindings(
+        request: SetWorkflowOutputBindingsRequest,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.set_workflow_output_bindings(
+                workspace_id=request.workspace_id,
+                revision=request.revision,
+                bindings=request.bindings,
+            )
+        )
+
     @server.tool(
         name="wf.workflow.set_workflow_output_map",
         title="Set Workflow Output Map",

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_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_bindings" in names
             assert "wf.workflow.set_workflow_output_map" in names
             assert "wf.workflow.bind" in names
             assert "wf.workflow.add_step_from_capability" in names
@@ -134,6 +135,14 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             ]
             assert "bindings" in set_output_bindings_request["properties"]
             assert "merge" not in set_output_bindings_request["properties"]
+            canonical_workflow_output_schema = tools_by_name[
+                "wf.workflow.set_workflow_output_bindings"
+            ].inputSchema
+            canonical_workflow_output_request = canonical_workflow_output_schema[
+                "properties"
+            ]["request"]
+            assert "bindings" in canonical_workflow_output_request["properties"]
+            assert "merge" not in canonical_workflow_output_request["properties"]
             set_workflow_output_schema = tools_by_name[
                 "wf.workflow.set_workflow_output_map"
             ].inputSchema

tests/wf_mcp/server/test_tools.py

@@ -9,7 +9,7 @@ from fastmcp.client import Client
 from fastmcp.client.transports.memory import FastMCPTransport
 
 from wf_artifacts import FileDraftWorkspaceStore, FileWorkflowArtifactStore
-from wf_core.models.steps import OutputBinding
+from wf_core.models.steps import InputPathBinding, InputValueBinding, OutputBinding
 from wf_mcp.broker import WfMcpService
 from wf_mcp.models import BrokerConfig
 from wf_mcp.server import create_server_client
@@ -52,6 +52,7 @@ async def test_server_search_mode_pins_stable_control_and_workflow_tools() -> No
         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_bindings" in names
         assert "wf.workflow.set_workflow_output_map" in names
         assert "wf.workflow.bind" in names
         assert "wf.workflow.remove_draft_route" in names
@@ -143,6 +144,74 @@ async def test_registered_output_bindings_tool_delegates_typed_bindings_once(
     ]
 
 
+@pytest.mark.asyncio
+async def test_registered_workflow_output_bindings_tool_preserves_union_order(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class RecordingWorkflowHandler:
+        def __init__(self) -> None:
+            self.calls: list[dict[str, Any]] = []
+
+        async def set_workflow_output_bindings(
+            self, **kwargs: Any
+        ) -> dict[str, Any]:
+            self.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 / "workflow_output_tool_store"),
+        artifact_store=FileWorkflowArtifactStore(
+            tmp_path / "workflow_output_tool_artifacts"
+        ),
+        draft_workspace_store=FileDraftWorkspaceStore(
+            tmp_path / "workflow_output_tool_drafts"
+        ),
+    )
+    server = FastMCP("workflow-output-bindings-test")
+    register_workflow_tools(server, service)
+
+    async with Client(FastMCPTransport(server)) as client:
+        result = await client.call_tool(
+            "wf.workflow.set_workflow_output_bindings",
+            {
+                "request": {
+                    "workspace_id": "draft-output",
+                    "revision": 4,
+                    "bindings": [
+                        {
+                            "path": "state.report.title",
+                            "target": "report.title",
+                        },
+                        {"value": "markdown", "target": "format"},
+                    ],
+                }
+            },
+        )
+
+    assert structured(result)["revision"] == 5
+    call = recorder.calls[0]
+    assert isinstance(call["bindings"][0], InputPathBinding)
+    assert isinstance(call["bindings"][1], InputValueBinding)
+    assert [
+        binding.model_dump(mode="json") for binding in call["bindings"]
+    ] == [
+        {"path": "state.report.title", "target": "report.title"},
+        {"value": "markdown", "target": "format"},
+    ]
+
+
 @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

@@ -21,6 +21,7 @@ from wf_mcp.workflow_surface.models import (
     CreateMinimalDraftWorkspaceRequest,
     SetStepInputBindingsRequest,
     SetStepOutputBindingsRequest,
+    SetWorkflowOutputBindingsRequest,
 )
 
 from ..test_support import echo_tool
@@ -332,6 +333,39 @@ def test_set_step_output_bindings_request_rejects_malformed_canonical_record() -
         )
 
 
+def test_set_workflow_output_bindings_request_preserves_union_order() -> None:
+    request = SetWorkflowOutputBindingsRequest.model_validate(
+        {
+            "workspace_id": "draft-output",
+            "revision": 4,
+            "bindings": [
+                {"path": "state.report.title", "target": "report.title"},
+                {"value": "markdown", "target": "format"},
+            ],
+        }
+    )
+
+    assert isinstance(request.bindings[0], InputPathBinding)
+    assert isinstance(request.bindings[1], InputValueBinding)
+    assert [
+        binding.model_dump(mode="json") for binding in request.bindings
+    ] == [
+        {"path": "state.report.title", "target": "report.title"},
+        {"value": "markdown", "target": "format"},
+    ]
+
+
+def test_set_workflow_output_bindings_request_rejects_malformed_record() -> None:
+    with pytest.raises(ValidationError):
+        SetWorkflowOutputBindingsRequest.model_validate(
+            {
+                "workspace_id": "draft-output",
+                "revision": 4,
+                "bindings": [{"target": "report.title"}],
+            }
+        )
+
+
 def test_workflow_surface_sets_ordered_canonical_step_output_bindings(
     tmp_path: Path,
 ) -> None: