migrate all everything to the new stuff

examples/authoring_control_flow.py

@@ -4,7 +4,18 @@ from typing import Any
 
 from pydantic import BaseModel
 
-from wf_authoring import NodeReturn, NodeSpec, WorkflowBuilder, node, outcome, state
+from wf_authoring import (
+    NodeReturn,
+    NodeSpec,
+    WorkflowBuilder,
+    input_from,
+    input_path,
+    node,
+    outcome,
+    output_to,
+    state,
+    state_path,
+)
 from wf_core import END
 
 
@@ -169,8 +180,8 @@ def _message_use(
     return graph.use(
         spec,
         id=id,
-        in_map={"state.message": "message"},
+        input=[input_from(state_path("message"), "message")],
-        out_map={"message": "state.message"},
+        output=[output_to("message", state_path("message"))],
     )
 
 
@@ -184,8 +195,8 @@ def _status_use(
     return graph.use(
         spec,
         id=id,
-        in_map={"state.status": "status"},
+        input=[input_from(state_path("status"), "status")],
-        out_map={"message": "state.message"},
+        output=[output_to("message", state_path("message"))],
     )
 
 
@@ -199,11 +210,11 @@ def _metrics_use(
     return graph.use(
         spec,
         id=id,
-        in_map={
+        input=[
-            "state.message": "message",
+            input_from(state_path("message"), "message"),
-            "state.length": "length",
+            input_from(state_path("length"), "length"),
-        },
+        ],
-        out_map={"message": "state.message"},
+        output=[output_to("message", state_path("message"))],
     )
 
 
@@ -213,8 +224,8 @@ def build_branch_workflow() -> WorkflowBuilder:
     router = graph.use(
         classify_message,
         id="classify",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={"message": "state.message"},
+        output=[output_to("message", state_path("message"))],
     )
     graph.branch(
         router,
@@ -237,8 +248,8 @@ def build_handle_workflow() -> WorkflowBuilder:
     lookup = graph.use(
         lookup_message,
         id="lookup",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={"message": "state.message"},
+        output=[output_to("message", state_path("message"))],
     )
     deliver = _message_use(graph, deliver_message, id="deliver")
     failed = _message_use(graph, fail_safely, id="failed")
@@ -256,8 +267,8 @@ def build_match_workflow() -> WorkflowBuilder:
     classifier = graph.use(
         classify_status,
         id="classify_status",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={"status": "state.status"},
+        output=[output_to("status", state_path("status"))],
     )
     decision = graph.match(
         state("status"),
@@ -282,11 +293,11 @@ def build_when_workflow() -> WorkflowBuilder:
     measure = graph.use(
         measure_text,
         id="measure",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={
+        output=[
-            "message": "state.message",
+            output_to("message", state_path("message")),
-            "length": "state.length",
+            output_to("length", state_path("length")),
-        },
+        ],
     )
     decision = graph.when(
         state("length").ge(6),
@@ -307,11 +318,11 @@ def build_choose_workflow() -> WorkflowBuilder:
     measure = graph.use(
         measure_text,
         id="measure",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={
+        output=[
-            "message": "state.message",
+            output_to("message", state_path("message")),
-            "length": "state.length",
+            output_to("length", state_path("length")),
-        },
+        ],
     )
     decision = graph.choose(
         (state("length").ge(20), _metrics_use(graph, long_message, id="long")),
@@ -333,8 +344,8 @@ def build_use_ref_workflow() -> WorkflowBuilder:
     echo = graph.use_ref(
         "demo.echo",
         id="echo",
-        in_map={"input.text": "message"},
+        input=[input_from(input_path("text"), "message")],
-        out_map={"echoed": "state.message"},
+        output=[output_to("echoed", state_path("message"))],
     )
     graph.connect(echo, "ok", END)
     graph.set_entry_point(echo)

examples/demo_workflow.py

@@ -9,7 +9,8 @@ DemoHandler = Callable[[dict[str, object], RuntimeContext], dict[str, object]]
 
 
 def build_demo_workflow() -> Workflow:
-    return Workflow.model_validate({
+    return Workflow.model_validate(
+        {
             "name": "drive_summary_demo",
             "input_schema": {
                 "type": "object",
@@ -120,8 +121,18 @@ def build_demo_workflow() -> Workflow:
                     "type": "node",
                     "node": "drive_list_files",
                     "desc": "List files from a Google Drive folder",
-                "in_map": {"input.folder_id": "folder_id"},
+                    "input": [
-                "out_map": {"documents": "state.documents"},
+                        {
+                            "target": {"root": "local", "parts": ["folder_id"]},
+                            "path": {"root": "input", "parts": ["folder_id"]},
+                        }
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["documents"]},
+                            "target": {"root": "state", "parts": ["documents"]},
+                        }
+                    ],
                 },
                 {
                     "id": "summarize_each",
@@ -136,16 +147,36 @@ def build_demo_workflow() -> Workflow:
                     "type": "node",
                     "node": "summarize_document",
                     "desc": "Summarize one document",
-                "in_map": {"context.document": "document"},
+                    "input": [
-                "out_map": {"item_summary": "state.item_summaries"},
+                        {
+                            "target": {"root": "local", "parts": ["document"]},
+                            "path": {"root": "context", "parts": ["document"]},
+                        }
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["item_summary"]},
+                            "target": {"root": "state", "parts": ["item_summaries"]},
+                        }
+                    ],
                 },
                 {
                     "id": "combine_summaries",
                     "type": "node",
                     "node": "combine_summaries",
                     "desc": "Combine item summaries into one final summary",
-                "in_map": {"state.item_summaries": "item_summaries"},
+                    "input": [
-                "out_map": {"summary": "state.summary"},
+                        {
+                            "target": {"root": "local", "parts": ["item_summaries"]},
+                            "path": {"root": "state", "parts": ["item_summaries"]},
+                        }
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["summary"]},
+                            "target": {"root": "state", "parts": ["summary"]},
+                        }
+                    ],
                 },
                 {
                     "id": "should_email",
@@ -161,8 +192,18 @@ def build_demo_workflow() -> Workflow:
                     "type": "node",
                     "node": "send_email",
                     "desc": "Send the summary by email",
-                "in_map": {"state.summary": "summary"},
+                    "input": [
-                "out_map": {"email_status": "state.email_status"},
+                        {
+                            "target": {"root": "local", "parts": ["summary"]},
+                            "path": {"root": "state", "parts": ["summary"]},
+                        }
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["email_status"]},
+                            "target": {"root": "state", "parts": ["email_status"]},
+                        }
+                    ],
                 },
                 {
                     "id": "approve_email",
@@ -183,7 +224,12 @@ def build_demo_workflow() -> Workflow:
                     "type": "node",
                     "node": "mark_email_skipped",
                     "desc": "Record that email delivery was skipped",
-                "out_map": {"email_status": "state.email_status"},
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["email_status"]},
+                            "target": {"root": "state", "parts": ["email_status"]},
+                        }
+                    ],
                 },
             ],
             "edges": [
@@ -203,7 +249,8 @@ def build_demo_workflow() -> Workflow:
                 {"from": "send_email", "outcome": "sent", "to": END},
                 {"from": "skip_email", "outcome": "ok", "to": END},
             ],
-    })
+        }
+    )
 
 
 def drive_list_files(

examples/mcp_tool_workflow.py

@@ -34,7 +34,8 @@ async def run_example() -> dict[str, object]:
 
     await service.refresh_connection_catalog("fixture.personal")
 
-    plan = RawWorkflowPlan.model_validate({
+    plan = RawWorkflowPlan.model_validate(
+        {
             "name": "mcp_echo_workflow",
             "input_schema": {
                 "type": "object",
@@ -56,15 +57,30 @@ async def run_example() -> dict[str, object]:
                     "id": "echo",
                     "type": "node",
                     "node": "fixture.personal.echo_tool",
-                "in_map": {"input.text": "text"},
+                    "input": [
-                "out_map": {"echoed": "state.echoed"},
+                        {
+                            "target": {"root": "local", "parts": ["text"]},
+                            "path": {"root": "input", "parts": ["text"]},
+                        }
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["echoed"]},
+                            "target": {"root": "state", "parts": ["echoed"]},
+                        }
+                    ],
                 },
                 {
                     "id": "raise_mcp_error",
                     "type": "node",
                     "node": "wf.std.runtime_error",
-                "in_map": {"input.text": "message"},
+                    "input": [
-                "out_map": {},
+                        {
+                            "target": {"root": "local", "parts": ["message"]},
+                            "path": {"root": "input", "parts": ["text"]},
+                        }
+                    ],
+                    "output": [],
                 },
             ],
             "edges": [
@@ -72,7 +88,8 @@ async def run_example() -> dict[str, object]:
                 {"from": "echo", "outcome": "error", "to": "raise_mcp_error"},
                 {"from": "raise_mcp_error", "outcome": "ok", "to": END},
             ],
-    })
+        }
+    )
 
     run = await service.run_workflow_from_plan(plan, {"text": "hello from MCP"})
     return {

examples/wrapper_normalization.py

@@ -4,7 +4,16 @@ from typing import Literal
 
 from pydantic import BaseModel
 
-from wf_authoring import NodeReturn, WorkflowBuilder, node, outcome
+from wf_authoring import (
+    NodeReturn,
+    WorkflowBuilder,
+    input_from,
+    input_path,
+    node,
+    outcome,
+    output_to,
+    state_path,
+)
 from wf_core import END
 
 
@@ -65,20 +74,20 @@ def build_normalized_wrapper() -> WorkflowBuilder:
     raw = graph.use(
         raw_status_tool,
         id="raw_tool",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={
+        output=[
-            "status": "state.status",
+            output_to("status", state_path("status")),
-            "message": "state.message",
+            output_to("message", state_path("message")),
-        },
+        ],
     )
     normalizer = graph.use(
         normalize_status,
         id="normalize",
-        in_map={
+        input=[
-            "state.status": "status",
+            input_from(state_path("status"), "status"),
-            "state.message": "message",
+            input_from(state_path("message"), "message"),
-        },
+        ],
-        out_map={"message": "state.message"},
+        output=[output_to("message", state_path("message"))],
     )
     graph.connect(raw, "ok", normalizer)
     graph.connect(normalizer, "done", END)

src/wf_authoring/__init__.py

@@ -10,9 +10,12 @@ from .dsl import (
     expr,
     graph_path,
     input,
+    input_from,
     input_path,
+    input_value,
     merge_maps,
     not_,
+    output_to,
     state,
     state_path,
 )
@@ -111,7 +114,9 @@ __all__ = [
     "first_item_or_none",
     "graph_path",
     "input",
+    "input_from",
     "input_path",
+    "input_value",
     "is_empty",
     "last_item",
     "last_item_or_none",
@@ -124,6 +129,7 @@ __all__ = [
     "node",
     "not_",
     "outcome",
+    "output_to",
     "reducer",
     "state",
     "state_field",

src/wf_authoring/dsl/__init__.py

@@ -9,7 +9,16 @@ from .conditions import (
     not_,
     state,
 )
-from .mapping import PathArg, bind_fields, bind_state, merge_maps, normalize_path
+from .mapping import (
+    PathArg,
+    bind_fields,
+    bind_state,
+    input_from,
+    input_value,
+    merge_maps,
+    normalize_path,
+    output_to,
+)
 from .paths import GraphPath, context_path, graph_path, input_path, state_path
 
 __all__ = [
@@ -26,10 +35,13 @@ __all__ = [
     "expr",
     "graph_path",
     "input",
+    "input_from",
     "input_path",
+    "input_value",
     "merge_maps",
     "normalize_path",
     "not_",
+    "output_to",
     "state",
     "state_path",
 ]

src/wf_authoring/dsl/mapping.py

@@ -1,10 +1,17 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
-from typing import TypeAlias
+from typing import Any, TypeAlias
 
+from wf_core.models.steps import InputPathBinding, InputValueBinding, OutputBinding
+
+from .path_inputs import (
+    PathInput,
+    coerce_graph_path,
+    coerce_local_path,
+    coerce_state_path,
+)
 from .paths import GraphPath
-from .path_inputs import PathInput, coerce_graph_path, coerce_state_path
 
 PathArg: TypeAlias = PathInput | GraphPath
 
@@ -41,6 +48,30 @@ def bind_state(**mapping: PathArg) -> dict[str, str]:
     }
 
 
+def input_from(path: PathArg, target: PathInput) -> InputPathBinding:
+    """Bind a workflow graph path into a node-local input path."""
+    return InputPathBinding(
+        target=coerce_local_path(target),
+        path=coerce_graph_path(path.path if isinstance(path, GraphPath) else path),
+    )
+
+
+def input_value(target: PathInput, value: Any) -> InputValueBinding:
+    """Bind a literal value into a node-local input path."""
+    return InputValueBinding(target=coerce_local_path(target), value=value)
+
+
+def output_to(source: PathInput, target: PathArg) -> OutputBinding:
+    """Bind a node-local output path back into workflow state."""
+    return OutputBinding(
+        source=coerce_local_path(source),
+        target=coerce_state_path(
+            target.path if isinstance(target, GraphPath) else target,
+            allow_legacy_root=True,
+        ),
+    )
+
+
 def merge_maps(*maps: Mapping[str, str]) -> dict[str, str]:
     merged: dict[str, str] = {}
     for mapping in maps:

tests/artifacts/test_factory.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 from typing import Any, cast
 
 from wf_artifacts import RequiredCapability, create_workflow_artifact_from_plan
+from wf_core.models.steps import InputPathBinding, OutputBinding
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 from wf_platform import NodeSpecInventory
 
 
@@ -265,8 +267,8 @@ def _plan() -> dict[str, object]:
                 "id": "echo",
                 "type": "node",
                 "node": "demo.echo_tool",
-                "in_map": {"input.text": "text"},
+                "input": [_input_binding("input.text", "text")],
-                "out_map": {"echoed": "state.echoed"},
+                "output": [_output_binding("echoed", "state.echoed")],
             }
         ],
         "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],
@@ -277,3 +279,19 @@ def _set_first_node_ref(plan: dict[str, object], node_ref: str) -> None:
     """Set the first node ref in a loosely typed raw plan test fixture."""
     nodes = cast("list[dict[str, Any]]", plan["nodes"])
     nodes[0]["node"] = node_ref
+
+
+def _input_binding(path: str, target: str) -> dict[str, object]:
+    """Return canonical JSON for raw workflow plan fixtures."""
+    return InputPathBinding(
+        path=GraphSourcePath.parse(path),
+        target=LocalPath.parse(target),
+    ).model_dump(mode="json")
+
+
+def _output_binding(source: str, target: str) -> dict[str, object]:
+    """Return canonical JSON for raw workflow plan fixtures."""
+    return OutputBinding(
+        source=LocalPath.parse(source),
+        target=StatePath.parse(target),
+    ).model_dump(mode="json")

tests/authoring/test_builder.py

@@ -5,7 +5,14 @@ from collections.abc import Iterator, Mapping
 
 import pytest
 
-from wf_authoring import WorkflowBuilder, input_path, state, state_path
+from wf_authoring import (
+    WorkflowBuilder,
+    input_from,
+    input_path,
+    output_to,
+    state,
+    state_path,
+)
 from wf_core import END, RunStatus, WorkflowExecutionError
 from wf_core.models.steps import InputPathBinding, InputValueBinding
 from wf_core.paths import GraphSourcePath, LocalPath, StatePath
@@ -59,8 +66,8 @@ def test_builder_preserves_explicit_nested_node_local_maps() -> None:
 
     step = builder.use(
         auto_bind_node,
-        in_map={"state.text": "payload.text"},
+        input=[input_from(state_path("text"), "payload.text")],
-        out_map={"payload.text": "state.text"},
+        output=[output_to("payload.text", state_path("text"))],
     )
 
     assert isinstance(step.input[0], InputPathBinding)
@@ -80,8 +87,8 @@ def test_builder_use_accepts_typed_paths_and_literal_iterable_paths() -> None:
 
     step = builder.use(
         auto_bind_node,
-        in_map={input_path('"text.with.dot"'): ("payload.text",)},
+        input=[input_from(input_path('"text.with.dot"'), ("payload.text",))],
-        out_map={("payload.text",): state_path(("state field",))},
+        output=[output_to(("payload.text",), state_path(("state field",)))],
     )
 
     assert isinstance(step.input[0], InputPathBinding)
@@ -139,8 +146,8 @@ def test_builder_preserves_explicit_root_node_local_maps() -> None:
 
     step = builder.use(
         auto_bind_node,
-        in_map={"state.text": "."},
+        input=[input_from(state_path("text"), ".")],
-        out_map={".": "state.text"},
+        output=[output_to(".", state_path("text"))],
     )
 
     assert isinstance(step.input[0], InputPathBinding)
@@ -161,8 +168,8 @@ def test_builder_emits_canonical_node_bindings() -> None:
     step = builder.use(
         auto_bind_node,
         id="update",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={"text": "state.text"},
+        output=[output_to("text", state_path("text"))],
     )
     builder.connect(step, "ok", END)
 
@@ -300,8 +307,8 @@ def test_builder_use_ref_creates_external_node_use_without_node_def() -> None:
     step = builder.use_ref(
         "demo.echo",
         id="echo",
-        in_map={"input.text": "text"},
+        input=[input_from(input_path("text"), "text")],
-        out_map={"echoed": "state.echoed"},
+        output=[output_to("echoed", state_path("echoed"))],
     )
     builder.set_entry_point(step)
     builder.connect(step, "ok", END)

tests/authoring/test_demo_workflow.py

@@ -21,7 +21,9 @@ from wf_authoring import (
     bind_fields,
     bind_state,
     build_registry,
+    input_from,
     state,
+    output_to,
     state_path,
     context_path,
     input_path,
@@ -151,8 +153,8 @@ def build_authoring_demo_workflow():
     list_files = builder.use(
         drive_list_files_spec,
         id="list_files",
-        in_map=bind_fields(folder_id=input_path("folder_id")),
+        input=[input_from(input_path("folder_id"), "folder_id")],
-        out_map=bind_state(documents=state_path("documents")),
+        output=[output_to("documents", state_path("documents"))],
         desc="List files from a Google Drive folder",
     )
     summarize_each = builder.foreach(
@@ -165,15 +167,15 @@ def build_authoring_demo_workflow():
     summarize_one = builder.use(
         summarize_document_spec,
         id="summarize_one",
-        in_map=bind_fields(document=context_path("document")),
+        input=[input_from(context_path("document"), "document")],
-        out_map=bind_state(item_summary=state_path("item_summaries")),
+        output=[output_to("item_summary", state_path("item_summaries"))],
         desc="Summarize one document",
     )
     combine_summaries = builder.use(
         combine_summaries_spec,
         id="combine_summaries",
-        in_map=bind_fields(item_summaries=state_path("item_summaries")),
+        input=[input_from(state_path("item_summaries"), "item_summaries")],
-        out_map=bind_state(summary=state_path("summary")),
+        output=[output_to("summary", state_path("summary"))],
         desc="Combine item summaries into one final summary",
     )
     should_email = builder.condition(
@@ -183,8 +185,8 @@ def build_authoring_demo_workflow():
     send_email = builder.use(
         send_email_spec,
         id="send_email",
-        in_map=bind_fields(summary=state_path("summary")),
+        input=[input_from(state_path("summary"), "summary")],
-        out_map=bind_state(email_status=state_path("email_status")),
+        output=[output_to("email_status", state_path("email_status"))],
         desc="Send the summary by email",
     )
     approve_email = builder.interrupt(
@@ -203,7 +205,7 @@ def build_authoring_demo_workflow():
     skip_email = builder.use(
         mark_email_skipped_spec,
         id="skip_email",
-        out_map=bind_state(email_status=state_path("email_status")),
+        output=[output_to("email_status", state_path("email_status"))],
         desc="Record that email delivery was skipped",
     )
 
@@ -373,12 +375,14 @@ def test_foreach_stress_with_many_documents() -> None:
     assert len(run.state["documents"]) == document_count
     assert len(run.state["item_summaries"]) == document_count
     assert (
-        len([
+        len(
+            [
                 frame
                 for frame in run.frames.values()
                 if frame.kind == "foreach_iteration"
                 and frame.status == FrameStatus.COMPLETED
-        ])
+            ]
+        )
         == document_count
     )
     assert len([entry for entry in run.trace if entry.step_type == "foreach"]) == (

tests/authoring/test_ops.py

@@ -4,8 +4,6 @@ import pytest
 
 from wf_authoring import (
     WorkflowBuilder,
-    bind_fields,
-    bind_state,
     build_registry,
     coalesce,
     constant,
@@ -14,9 +12,11 @@ from wf_authoring import (
     first_item_maybe,
     first_item_or_none,
     is_empty,
+    input_from,
     last_item,
     last_item_or_none,
     length,
+    output_to,
     pick_path,
     pick_key,
     project_fields,
@@ -39,21 +39,23 @@ def _build_first_workflow(use_safe_first: bool = False):
     builder = WorkflowBuilder(
         name="first_demo",
         input_schema=SchemaRef(type="object"),
-        state_schema=StateSchema.model_validate({
+        state_schema=StateSchema.model_validate(
+            {
                 "type": "object",
                 "properties": {
                     "items": {"type": "array"},
                     "item": {"type": ["string", "null"]},
                 },
-        }),
+            }
+        ),
         output_schema=SchemaRef(type="object"),
         start="pick_first",
     )
     node = builder.use(
         spec,
         id="pick_first",
-        in_map=bind_fields(items=state_path("items")),
+        input=[input_from(state_path("items"), "items")],
-        out_map=bind_state(item=state_path("item")),
+        output=[output_to("item", state_path("item"))],
     )
     builder.connect(node, "ok", "__end__")
     return builder.compile(), build_registry(spec)
@@ -63,28 +65,30 @@ def _build_first_maybe_workflow():
     builder = WorkflowBuilder(
         name="first_maybe_demo",
         input_schema=SchemaRef(type="object"),
-        state_schema=StateSchema.model_validate({
+        state_schema=StateSchema.model_validate(
+            {
                 "type": "object",
                 "properties": {
                     "items": {"type": "array"},
                     "item": {"type": ["string", "null"]},
                     "missing": {"type": "boolean"},
                 },
-        }),
+            }
+        ),
         output_schema=SchemaRef(type="object"),
         start="pick_first",
     )
     pick_first = builder.use(
         first_item_maybe,
         id="pick_first",
-        in_map=bind_fields(items=state_path("items")),
+        input=[input_from(state_path("items"), "items")],
-        out_map=bind_state(item=state_path("item")),
+        output=[output_to("item", state_path("item"))],
     )
     mark_missing = builder.use(
         first_item_or_none,
         id="mark_missing",
-        in_map=bind_fields(items=state_path("items")),
+        input=[input_from(state_path("items"), "items")],
-        out_map=bind_state(item=state_path("item")),
+        output=[output_to("item", state_path("item"))],
     )
     builder.connect(pick_first, "found", "__end__")
     builder.connect(pick_first, "missing", mark_missing)

tests/authoring/test_reducers.py

@@ -9,8 +9,10 @@ from wf_authoring import (
     ReducerCatalog,
     WorkflowBuilder,
     node,
+    output_to,
     reducer,
     state_field,
+    state_path,
 )
 from wf_core import ReducerRef
 
@@ -102,8 +104,7 @@ def test_builder_executes_with_custom_reducer_catalog() -> None:
     )
     step = builder.use(
         emit,
-        in_map={},
+        output=[output_to("total", state_path("total"))],
-        out_map={"total": "state.total"},
     )
     builder.set_entry_point(step)
     builder.connect(step, "ok", "__end__")

tests/wf_mcp/test_broker_server.py

@@ -23,7 +23,9 @@ from .test_support import (
     FailingDiscoveryAdapter,
     FakeAdapter,
     echo_tool,
+    input_binding,
     local_temp_root,
+    output_binding,
 )
 
 
@@ -32,7 +34,8 @@ def test_load_broker_config_resolves_relative_store_root() -> None:
     tmp_path.mkdir(parents=True, exist_ok=True)
     config_path = tmp_path / "wf_mcp.config.json"
     config_path.write_text(
-        json.dumps({
+        json.dumps(
+            {
                 "store_root": ".broker-store",
                 "connections": [
                     {
@@ -41,7 +44,8 @@ def test_load_broker_config_resolves_relative_store_root() -> None:
                         "account": "personal",
                     }
                 ],
-        }),
+            }
+        ),
         encoding="utf-8",
     )
 
@@ -518,8 +522,8 @@ def _echo_artifact() -> WorkflowArtifact:
                     "id": "echo",
                     "type": "node",
                     "node": "demo.personal.echo_tool",
-                    "in_map": {"input.text": "text"},
+                    "input": [input_binding("input.text", "text")],
-                    "out_map": {"echoed": "state.echoed"},
+                    "output": [output_binding("echoed", "state.echoed")],
                 }
             ],
             "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],

tests/wf_mcp/test_service.py

@@ -26,7 +26,9 @@ from .test_support import (
     FakeAdapter,
     echo_tool,
     finalize_tool,
+    input_binding,
     local_temp_root,
+    output_binding,
 )
 
 
@@ -55,8 +57,8 @@ def _single_echo_plan(plan_name: str, node_name: str) -> RawWorkflowPlan:
                 "id": "echo",
                 "type": "node",
                 "node": node_name,
-                "in_map": {"input.text": "text"},
+                "input": [input_binding("input.text", "text")],
-                "out_map": {"echoed": "state.echoed"},
+                "output": [output_binding("echoed", "state.echoed")],
             }
         ],
         edges=[
@@ -360,15 +362,15 @@ def test_service_compiles_and_runs_raw_plan() -> None:
                 "id": "echo",
                 "type": "node",
                 "node": "demo.personal.echo_tool",
-                "in_map": {"input.text": "text"},
+                "input": [input_binding("input.text", "text")],
-                "out_map": {"echoed": "state.echoed"},
+                "output": [output_binding("echoed", "state.echoed")],
             },
             {
                 "id": "finalize",
                 "type": "node",
                 "node": "demo.personal.finalize_tool",
-                "in_map": {"state.echoed": "echoed"},
+                "input": [input_binding("state.echoed", "echoed")],
-                "out_map": {"result": "state.result"},
+                "output": [output_binding("result", "state.result")],
             },
         ],
         edges=[
@@ -424,8 +426,8 @@ def test_service_resolves_registered_spec_with_dotted_local_name() -> None:
                 "id": "echo",
                 "type": "node",
                 "node": "demo.personal.foo.bar",
-                "in_map": {"input.text": "text"},
+                "input": [input_binding("input.text", "text")],
-                "out_map": {"echoed": "state.echoed"},
+                "output": [output_binding("echoed", "state.echoed")],
             }
         ],
         edges=[
@@ -571,8 +573,8 @@ def test_service_does_not_resolve_specs_hidden_from_planner() -> None:
                 "id": "echo",
                 "type": "node",
                 "node": "hidden.source.echo_tool",
-                "in_map": {"input.text": "text"},
+                "input": [input_binding("input.text", "text")],
-                "out_map": {"echoed": "state.echoed"},
+                "output": [output_binding("echoed", "state.echoed")],
             }
         ],
         edges=[
@@ -865,8 +867,8 @@ def test_service_records_tool_call_events() -> None:
                 "id": "echo",
                 "type": "node",
                 "node": "demo.personal.echo_tool",
-                "in_map": {"input.text": "text"},
+                "input": [input_binding("input.text", "text")],
-                "out_map": {"echoed": "state.echoed"},
+                "output": [output_binding("echoed", "state.echoed")],
             }
         ],
         edges=[

tests/wf_mcp/test_support.py

@@ -9,6 +9,8 @@ from pydantic import BaseModel, Field
 
 from wf_authoring import NodeReturn, node
 from wf_core import RuntimeContext
+from wf_core.models.steps import InputPathBinding, OutputBinding
+from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 from wf_mcp.capabilities import DiscoveredPrompt, DiscoveredResource, DiscoveredTool
 from wf_mcp.models import AuthRecord, ConnectionConfig
 from wf_mcp.sdk import ToolCallResult
@@ -55,6 +57,22 @@ def fixture_server_path() -> str:
     return str(Path(__file__).resolve().parents[1] / "fixtures" / "mcp_echo_server.py")
 
 
+def input_binding(path: str, target: str) -> dict[str, object]:
+    """Return canonical JSON for a node input path binding in raw plans."""
+    return InputPathBinding(
+        path=GraphSourcePath.parse(path),
+        target=LocalPath.parse(target),
+    ).model_dump(mode="json")
+
+
+def output_binding(source: str, target: str) -> dict[str, object]:
+    """Return canonical JSON for a node output binding in raw plans."""
+    return OutputBinding(
+        source=LocalPath.parse(source),
+        target=StatePath.parse(target),
+    ).model_dump(mode="json")
+
+
 def everything_server_connection() -> ConnectionConfig | None:
     transport = os.environ.get("MCP_EVERYTHING_TRANSPORT", "stdio")
     if transport == "stdio":

tests/wf_mcp/test_workflow_surface.py

@@ -25,7 +25,7 @@ from wf_platform import (
     SourceVisibility,
 )
 
-from .test_support import echo_tool, local_temp_root
+from .test_support import echo_tool, input_binding, local_temp_root, output_binding
 
 
 class AmountInput(BaseModel):
@@ -1263,8 +1263,8 @@ def _echo_artifact() -> WorkflowArtifact:
                 "id": "echo",
                 "type": "node",
                 "node": "demo.personal.echo_tool",
-                "in_map": {"input.text": "text"},
+                "input": [input_binding("input.text", "text")],
-                "out_map": {"echoed": "state.echoed"},
+                "output": [output_binding("echoed", "state.echoed")],
             }
         ],
         "edges": [{"from": "echo", "outcome": "ok", "to": "__end__"}],
@@ -1367,8 +1367,8 @@ def _custom_reducer_artifact() -> WorkflowArtifact:
                 "id": "amount",
                 "type": "node",
                 "node": "demo.personal.amount_tool",
-                "in_map": {"input.amount": "amount"},
+                "input": [input_binding("input.amount", "amount")],
-                "out_map": {"amount": "state.total"},
+                "output": [output_binding("amount", "state.total")],
             }
         ],
         "edges": [{"from": "amount", "outcome": "ok", "to": "__end__"}],