draft: ergonomic workflow building for LLM

src/wf_artifacts/__init__.py

@@ -4,6 +4,11 @@ from .catalog import (
     artifact_node_name,
 )
 from .factory import create_workflow_artifact_from_plan
+from .drafts import (
+    compile_workflow_draft,
+    patch_workflow_draft,
+    validate_workflow_draft,
+)
 from .models import (
     ArtifactKind,
     AvailableCapability,
@@ -37,7 +42,10 @@ __all__ = [
     "artifact_catalog_entry",
     "artifact_node_name",
     "create_workflow_artifact_from_plan",
+    "compile_workflow_draft",
     "logical_ref_for_concrete_ref",
     "normalize_plan_node_refs",
+    "patch_workflow_draft",
     "validate_deployment_dependencies",
+    "validate_workflow_draft",
 ]

src/wf_artifacts/drafts.py

@@ -0,0 +1,269 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+import jsonpatch
+from pydantic import ValidationError
+
+from wf_core import Workflow
+
+JsonObject = dict[str, Any]
+JsonPatch = list[dict[str, Any]]
+
+
+def compile_workflow_draft(draft: JsonObject) -> JsonObject:
+    """Compile the LLM-friendly draft shape into the normalized workflow plan."""
+    plan = _compile_unvalidated_draft(draft)
+    _validate_plan_model(plan)
+    _validate_graph_references(plan)
+    return plan
+
+
+def validate_workflow_draft(draft: JsonObject) -> JsonObject:
+    """Return structured draft diagnostics instead of raising on bad input."""
+    try:
+        compiled_plan = compile_workflow_draft(draft)
+    except Exception as exc:
+        return {
+            "status": "invalid",
+            "diagnostics": [_diagnostic_from_exception(exc)],
+        }
+    return {
+        "status": "valid",
+        "diagnostics": [],
+        "compiled_plan": compiled_plan,
+    }
+
+
+def patch_workflow_draft(draft: JsonObject, patch: JsonPatch) -> JsonObject:
+    """Apply RFC 6902 JSON Patch to a draft, then validate the patched draft.
+
+    Patch authoring is intentionally draft-first. Compiled raw plans are compiler
+    output, so callers should patch the readable source document and recompile it.
+    """
+    try:
+        patched = jsonpatch.JsonPatch(patch).apply(deepcopy(draft), in_place=False)
+    except Exception as exc:
+        return {
+            "status": "invalid",
+            "diagnostics": [
+                {
+                    "code": "patch_invalid",
+                    "path": "patch",
+                    "message": str(exc),
+                }
+            ],
+        }
+    if not isinstance(patched, dict):
+        return {
+            "status": "invalid",
+            "diagnostics": [
+                {
+                    "code": "draft_not_object",
+                    "path": "",
+                    "message": "patched draft must be a JSON object",
+                }
+            ],
+        }
+    result = validate_workflow_draft(patched)
+    return {"draft": patched, **result}
+
+
+def _compile_unvalidated_draft(draft: JsonObject) -> JsonObject:
+    if not isinstance(draft, dict):
+        raise ValueError("draft must be a JSON object")
+
+    plan: JsonObject = {
+        "name": _required_str(draft, "name"),
+        "input_schema": _required_object(draft, "input_schema"),
+        "state_schema": _required_object(draft, "state_schema"),
+        "output_schema": _required_object(draft, "output_schema"),
+        "start": _required_str(draft, "start"),
+        "nodes": [
+            _compile_step(step, index) for index, step in enumerate(_steps(draft))
+        ],
+        "edges": [
+            _compile_edge(edge, index) for index, edge in enumerate(_edges(draft))
+        ],
+    }
+    return plan
+
+
+def _compile_step(step: object, index: int) -> JsonObject:
+    path = f"steps[{index}]"
+    if not isinstance(step, dict):
+        raise ValueError(f"{path} must be a JSON object")
+
+    step_id = _required_str(step, "id", path=path)
+    kind = _required_str(step, "kind", path=path)
+    if kind == "use":
+        node: JsonObject = {
+            "id": step_id,
+            "type": "node",
+            "node": _required_str(step, "capability", path=path),
+            "in_map": _optional_object(step, "in", default={}),
+            "out_map": _optional_object(step, "out", default={}),
+        }
+        _copy_optional(step, node, "desc")
+        _copy_optional(step, node, "retry")
+        _copy_optional(step, node, "timeout_seconds")
+        return node
+    if kind == "condition":
+        return {
+            "id": step_id,
+            "type": "condition",
+            "check": _required_object(step, "check", path=path),
+        }
+    if kind == "foreach":
+        node = {
+            "id": step_id,
+            "type": "foreach",
+            "over": _required_str(step, "over", path=path),
+            "as": _required_str(step, "as", path=path),
+        }
+        _copy_optional(step, node, "mode")
+        _copy_optional(step, node, "on_item_error")
+        return node
+    if kind == "interrupt":
+        node = {
+            "id": step_id,
+            "type": "interrupt",
+            "kind": _required_str(step, "interrupt_kind", path=path),
+            "request_map": _optional_object(step, "request", default={}),
+            "out_map": _optional_object(step, "resume", default={}),
+        }
+        _copy_optional(step, node, "outcomes")
+        return node
+    if kind == "join":
+        return {"id": step_id, "type": "join"}
+    raise ValueError(f"{path}.kind has unsupported value {kind!r}")
+
+
+def _compile_edge(edge: object, index: int) -> JsonObject:
+    path = f"edges[{index}]"
+    if not isinstance(edge, dict):
+        raise ValueError(f"{path} must be a JSON object")
+    return {
+        "from": _required_str(edge, "from", path=path),
+        "outcome": _required_str(edge, "outcome", path=path),
+        "to": _required_str(edge, "to", path=path),
+    }
+
+
+def _validate_plan_model(plan: JsonObject) -> None:
+    try:
+        Workflow.model_validate(plan)
+    except ValidationError as exc:
+        raise ValueError(_validation_error_message(exc)) from exc
+
+
+def _validate_graph_references(plan: JsonObject) -> None:
+    node_ids = [node["id"] for node in plan["nodes"] if isinstance(node, dict)]
+    node_id_set = set(node_ids)
+    if len(node_ids) != len(node_id_set):
+        raise ValueError("steps contain duplicate ids")
+    if plan["start"] not in node_id_set:
+        raise ValueError(f"start references unknown step id {plan['start']!r}")
+    for index, edge in enumerate(plan["edges"]):
+        if edge["from"] not in node_id_set:
+            raise ValueError(
+                f"edges[{index}].from references unknown step id {edge['from']!r}"
+            )
+        if edge["to"] != "__end__" and edge["to"] not in node_id_set:
+            raise ValueError(
+                f"edges[{index}].to references unknown step id {edge['to']!r}"
+            )
+
+
+def _steps(draft: JsonObject) -> list[object]:
+    steps = draft.get("steps")
+    if not isinstance(steps, list):
+        raise ValueError("steps must be an array")
+    return steps
+
+
+def _edges(draft: JsonObject) -> list[object]:
+    edges = draft.get("edges")
+    if not isinstance(edges, list):
+        raise ValueError("edges must be an array")
+    return edges
+
+
+def _required_object(payload: JsonObject, key: str, *, path: str = "") -> JsonObject:
+    value = payload.get(key)
+    if not isinstance(value, dict):
+        raise ValueError(f"{_join_path(path, key)} must be an object")
+    return deepcopy(value)
+
+
+def _optional_object(
+    payload: JsonObject,
+    key: str,
+    *,
+    default: JsonObject,
+) -> JsonObject:
+    value = payload.get(key, default)
+    if not isinstance(value, dict):
+        raise ValueError(f"{key} must be an object")
+    return deepcopy(value)
+
+
+def _required_str(payload: JsonObject, key: str, *, path: str = "") -> str:
+    value = payload.get(key)
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"{_join_path(path, key)} must be a non-empty string")
+    return value
+
+
+def _copy_optional(source: JsonObject, target: JsonObject, key: str) -> None:
+    if key in source:
+        target[key] = deepcopy(source[key])
+
+
+def _join_path(path: str, key: str) -> str:
+    return f"{path}.{key}" if path else key
+
+
+def _validation_error_message(exc: ValidationError) -> str:
+    first_error = exc.errors()[0]
+    location = ".".join(str(part) for part in first_error["loc"])
+    return f"{_draft_path(location)}: {first_error['msg']}"
+
+
+def _diagnostic_from_exception(exc: Exception) -> JsonObject:
+    message = str(exc)
+    path = _path_from_message(message)
+    return {
+        "code": "draft_invalid",
+        "path": path,
+        "step_id": _step_id_from_path(path),
+        "message": message,
+    }
+
+
+def _path_from_message(message: str) -> str:
+    if ":" in message and message.split(":", 1)[0]:
+        return _draft_path(message.split(":", 1)[0])
+    token = message.split(" ", 1)[0]
+    if token.startswith(("steps[", "edges[")) or token in {
+        "name",
+        "input_schema",
+        "state_schema",
+        "output_schema",
+        "start",
+        "steps",
+        "edges",
+    }:
+        return _draft_path(token)
+    return ""
+
+
+def _draft_path(path: str) -> str:
+    return path.replace("nodes[", "steps[").replace(".type", ".kind")
+
+
+def _step_id_from_path(path: str) -> str | None:
+    if not path.startswith("steps["):
+        return None
+    return None

src/wf_mcp/broker/discovery.py

@@ -18,6 +18,7 @@ from .events import McpEvent
 
 _CapabilityT = TypeVar("_CapabilityT")
 
+
 @dataclass(slots=True)
 class DiscoveredConnectionCapabilities:
     tools: list[DiscoveredTool] = field(default_factory=list)

src/wf_mcp/transparent_proxy/runtime.py

@@ -36,7 +36,11 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     # Stable workflow control surface. Keep future workflow-capability test
     # tools pinned here too; they are distinct from raw MCP tool execution.
     "wf.workflow.list_artifacts",
+    "wf.workflow.validate_draft",
+    "wf.workflow.compile_draft",
     "wf.workflow.create_artifact_from_plan",
+    "wf.workflow.create_artifact_from_draft",
+    "wf.workflow.patch_draft",
     "wf.workflow.call_capability",
     "wf.workflow.inspect_artifact",
     "wf.workflow.list_deployments",

src/wf_mcp/workflow_surface/handlers.py

@@ -13,7 +13,10 @@ from wf_artifacts import (
     WorkflowArtifact,
     WorkflowCapabilityRef,
     WorkflowDeployment,
+    compile_workflow_draft,
     create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
+    patch_workflow_draft,
+    validate_workflow_draft,
     validate_deployment_dependencies,
 )
 from wf_platform import CapabilityRef, NodeSpecInventory, hash_json_schema, page_items
@@ -250,6 +253,89 @@ class WorkflowSurfaceHandlers:
             "saved": True,
         }
 
+    async def validate_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
+        return validate_workflow_draft(draft)
+
+    async def compile_draft(self, *, draft: dict[str, Any]) -> dict[str, Any]:
+        plan = compile_workflow_draft(draft)
+        return {
+            "compiled_plan": plan,
+            "required_capabilities": _required_capability_payloads(
+                _required_capabilities_for_plan(
+                    plan,
+                    source_bindings=None,
+                    service=self.service,
+                )
+            ),
+        }
+
+    async def create_artifact_from_draft(
+        self,
+        *,
+        artifact_id: str,
+        version: int,
+        title: str,
+        draft: dict[str, Any],
+        outcomes: Sequence[str],
+        kind: ArtifactKind = "workflow",
+        description: str | None = None,
+        required_capabilities: dict[str, dict[str, Any]] | None = None,
+        source_bindings: dict[str, str] | None = None,
+        created_from_catalog_version: str | None = None,
+    ) -> dict[str, Any]:
+        if self.service.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+        plan = compile_workflow_draft(draft)
+        workflow_artifact = build_workflow_artifact_from_plan(
+            artifact_id=artifact_id,
+            version=version,
+            title=title,
+            kind=kind,
+            description=description,
+            plan=plan,
+            outcomes=tuple(outcomes),
+            required_capabilities={
+                name: RequiredCapability.model_validate(capability)
+                for name, capability in (required_capabilities or {}).items()
+            },
+            source_bindings=source_bindings,
+            observed_node_specs=_observed_node_specs(self.service),
+            created_from_catalog_version=created_from_catalog_version,
+        )
+        self.service.artifact_store.save_artifact(workflow_artifact)
+        self.service._record_event(
+            make_event(
+                "workflow_artifact_saved",
+                capability_id=_artifact_capability_id(workflow_artifact),
+                payload={
+                    "artifact_id": workflow_artifact.id,
+                    "version": workflow_artifact.version,
+                    "created_from_draft": True,
+                },
+            )
+        )
+        required_sources = sorted(
+            {
+                capability.logical_source
+                for capability in workflow_artifact.required_capabilities.values()
+            }
+        )
+        return {
+            "artifact_id": workflow_artifact.id,
+            "version": workflow_artifact.version,
+            "saved": True,
+            "required_logical_sources": required_sources,
+            "suggested_bindings": _suggested_self_bindings(required_sources),
+        }
+
+    async def patch_draft(
+        self,
+        *,
+        draft: dict[str, Any],
+        patch: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        return patch_workflow_draft(draft, patch)
+
     async def inspect_artifact(
         self, *, artifact_id: str, version: int
     ) -> dict[str, Any]:
@@ -400,16 +486,62 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
     return sources
 
 
+def _required_capabilities_for_plan(
+    plan: dict[str, Any],
+    *,
+    source_bindings: dict[str, str] | None,
+    service: WfMcpService,
+) -> dict[str, RequiredCapability]:
+    """Infer a draft dependency summary without persisting an artifact."""
+    artifact = build_workflow_artifact_from_plan(
+        artifact_id="draft_preview",
+        version=1,
+        title="Draft Preview",
+        plan=plan,
+        outcomes=("completed",),
+        source_bindings=source_bindings,
+        observed_node_specs=_observed_node_specs(service),
+    )
+    requirements = dict(artifact.required_capabilities)
+    for node in _plan_nodes(artifact):
+        raw_ref = node.get("node")
+        if not isinstance(raw_ref, str) or raw_ref in requirements:
+            continue
+        try:
+            parsed = CapabilityRef.parse(raw_ref)
+        except ValueError:
+            continue
+        requirements[raw_ref] = RequiredCapability(
+            logical_source=str(parsed.source),
+            capability_name=parsed.name,
+            kind="node_spec",
+        )
+    return requirements
+
+
+def _required_capability_payloads(
+    requirements: dict[str, RequiredCapability],
+) -> dict[str, dict[str, Any]]:
+    return {
+        name: capability.model_dump(mode="json")
+        for name, capability in sorted(requirements.items())
+    }
+
+
+def _suggested_self_bindings(required_sources: Sequence[str]) -> dict[str, str]:
+    """Suggest local bindings for built-in sources that deploy to themselves."""
+    return {
+        source: source for source in required_sources if source in {"wf.std", "wf.mcp"}
+    }
+
+
 def _observed_node_specs(service: WfMcpService) -> dict[str, NodeSpecInventory]:
     """Project current executable specs into serializable observed contracts."""
     observed: dict[str, NodeSpecInventory] = {}
     for source in service.capability_sources.values():
         inventory = source.as_inventory()
         observed.update(
-            {
-                detail.name: detail
-                for detail in inventory.capabilities.node_spec_details
-            }
+            {detail.name: detail for detail in inventory.capabilities.node_spec_details}
         )
     return observed
 

src/wf_mcp/workflow_surface/tools.py

@@ -77,6 +77,22 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
     async def save_artifact(artifact: dict[str, Any]) -> dict[str, Any]:
         return await handlers.save_artifact(artifact)
 
+    @server.tool(
+        name="wf.workflow.validate_draft",
+        title="Validate Workflow Draft",
+        description="Validate an LLM-friendly workflow draft without saving it.",
+    )
+    async def validate_draft(draft: dict[str, Any]) -> dict[str, Any]:
+        return await handlers.validate_draft(draft=draft)
+
+    @server.tool(
+        name="wf.workflow.compile_draft",
+        title="Compile Workflow Draft",
+        description="Compile an LLM-friendly workflow draft into a raw workflow plan.",
+    )
+    async def compile_draft(draft: dict[str, Any]) -> dict[str, Any]:
+        return await handlers.compile_draft(draft=draft)
+
     @server.tool(
         name="wf.workflow.create_artifact_from_plan",
         title="Create Workflow Artifact From Plan",
@@ -117,6 +133,60 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             created_from_catalog_version=created_from_catalog_version,
         )
 
+    @server.tool(
+        name="wf.workflow.create_artifact_from_draft",
+        title="Create Workflow Artifact From Draft",
+        description=(
+            "Compile an LLM-friendly workflow draft and save it as a versioned "
+            "artifact."
+        ),
+    )
+    async def create_artifact_from_draft(
+        artifact_id: str,
+        version: int,
+        title: str,
+        draft: dict[str, Any],
+        outcomes: list[str],
+        kind: ArtifactKind = "workflow",
+        description: str | None = None,
+        required_capabilities: (
+            Mapping[str, RequiredCapability | dict[str, Any]] | None
+        ) = None,
+        source_bindings: Mapping[str, str] | None = None,
+        created_from_catalog_version: str | None = None,
+    ) -> dict[str, Any]:
+        return await handlers.create_artifact_from_draft(
+            artifact_id=artifact_id,
+            version=version,
+            title=title,
+            kind=kind,
+            description=description,
+            draft=draft,
+            outcomes=outcomes,
+            required_capabilities={
+                name: (
+                    capability.model_dump()
+                    if isinstance(capability, RequiredCapability)
+                    else capability
+                )
+                for name, capability in (required_capabilities or {}).items()
+            }
+            or None,
+            source_bindings=dict(source_bindings or {}),
+            created_from_catalog_version=created_from_catalog_version,
+        )
+
+    @server.tool(
+        name="wf.workflow.patch_draft",
+        title="Patch Workflow Draft",
+        description="Apply an RFC 6902 JSON Patch to a workflow draft and validate it.",
+    )
+    async def patch_draft(
+        draft: dict[str, Any],
+        patch: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        return await handlers.patch_draft(draft=draft, patch=patch)
+
     @server.tool(
         name="wf.workflow.inspect_artifact",
         title="Inspect Workflow Artifact",