draft workspace: the promised wf_authoring for LLMs

docs/wf_mcp_end_to_end_runbook.md

@@ -483,3 +483,15 @@ wf.workflow.validate_deployment
 - [`workflow_drafts.md`](workflow_drafts.md) for the preferred authoring format
 - [`workflow_artifacts.md`](workflow_artifacts.md) for immutable artifacts,
   deployments, and dependency contracts
+
+## Workspace Variant
+
+If the client is iterating with an LLM, prefer a draft workspace:
+
+1. Create a minimal workspace from the selected capability.
+2. Fetch the workspace by id when context is needed.
+3. Patch it by id and revision.
+4. Save an artifact from the workspace after validation is clean.
+
+This avoids resending the whole draft object every turn. The saved artifact is
+still immutable and should be deployed through the normal deployment path.

docs/wf_mcp_operator_manual.md

@@ -340,3 +340,19 @@ accounts.
 - [`workflow_drafts.md`](workflow_drafts.md) for the preferred authoring format
 - [`workflow_artifacts.md`](workflow_artifacts.md) for immutable artifacts,
   deployments, and saved workflows as future nodes
+
+## Draft Workspace Authoring
+
+Use draft workspaces when a client should iteratively edit one workflow without
+resending the full draft each turn.
+
+| Need | Tool |
+| --- | --- |
+| Start a patchable authoring session | `wf.workflow.create_minimal_draft_workspace` |
+| Fetch current draft workspace | `wf.workflow.get_draft_workspace` |
+| Patch current draft workspace | `wf.workflow.patch_draft_workspace` |
+| Save final workspace as artifact | `wf.workflow.create_artifact_from_workspace` |
+
+Workspace patches are optimistic-concurrency guarded. Pass the current
+`revision` from `get_draft_workspace`; a stale revision returns
+`revision_conflict` and leaves the stored draft unchanged.

docs/workflow_drafts.md

@@ -197,6 +197,29 @@ The workflow MCP surface exposes these draft tools:
 Use `validate_draft` before saving. Use `patch_draft` when an LLM client needs a
 small targeted correction instead of rewriting the whole workflow.
 
+## Draft Workspaces
+
+Stateless draft tools require the caller to resend the whole draft. Draft
+workspaces are the preferred LLM authoring flow when a client will patch a
+workflow over several turns.
+
+The workspace flow is:
+
+1. `wf.workflow.create_minimal_draft_workspace`
+2. `wf.workflow.get_draft_workspace`
+3. `wf.workflow.patch_draft_workspace`
+4. repeat get/patch until valid
+5. `wf.workflow.create_artifact_from_workspace`
+
+Workspaces are mutable and revisioned. Artifacts are immutable and versioned.
+Patch calls must include the current `revision`; stale revisions return
+`revision_conflict` and do not mutate the workspace.
+
+`create_minimal_draft_workspace` is intentionally only a bootstrapper. It wires
+an `error` outcome for naive MCP wrappers only when `error_message_source` is
+provided or a state path can be derived from `output_map`. Provider-specific
+error envelopes still belong in saved wrapper artifacts or follow-up patches.
+
 ## Patching Drafts
 
 `patch_draft` accepts JSON Patch operations.

src/wf_artifacts/__init__.py

@@ -9,6 +9,17 @@ from .drafts import (
     patch_workflow_draft,
     validate_workflow_draft,
 )
+from .draft_workspaces import (
+    DraftWorkspaceStore,
+    DraftWorkspaceConflictError,
+    FileDraftWorkspaceStore,
+    WorkflowDraftWorkspace,
+    create_draft_workspace,
+    ensure_workspace_id,
+    get_draft_workspace,
+    patch_draft_workspace,
+    summarize_draft_workspace,
+)
 from .models import (
     ArtifactKind,
     AvailableCapability,
@@ -32,20 +43,29 @@ __all__ = [
     "DependencyDiagnostic",
     "DiagnosticSeverity",
     "DriftPolicy",
+    "DraftWorkspaceConflictError",
+    "DraftWorkspaceStore",
+    "FileDraftWorkspaceStore",
     "FileWorkflowArtifactStore",
     "RequiredCapability",
     "WorkflowArtifact",
     "WorkflowArtifactCatalogEntry",
     "WorkflowCapabilityRef",
+    "WorkflowDraftWorkspace",
     "WorkflowArtifactStore",
     "WorkflowDeployment",
     "artifact_catalog_entry",
     "artifact_node_name",
+    "create_draft_workspace",
     "create_workflow_artifact_from_plan",
     "compile_workflow_draft",
+    "ensure_workspace_id",
+    "get_draft_workspace",
     "logical_ref_for_concrete_ref",
     "normalize_plan_node_refs",
+    "patch_draft_workspace",
     "patch_workflow_draft",
+    "summarize_draft_workspace",
     "validate_deployment_dependencies",
     "validate_workflow_draft",
 ]

src/wf_artifacts/draft_workspaces/__init__.py

@@ -0,0 +1,23 @@
+from .api import create_draft_workspace, get_draft_workspace, patch_draft_workspace
+from .models import (
+    WorkflowDraftWorkspace,
+    ensure_workspace_id,
+    summarize_draft_workspace,
+)
+from .store import (
+    DraftWorkspaceConflictError,
+    DraftWorkspaceStore,
+    FileDraftWorkspaceStore,
+)
+
+__all__ = [
+    "DraftWorkspaceConflictError",
+    "DraftWorkspaceStore",
+    "FileDraftWorkspaceStore",
+    "WorkflowDraftWorkspace",
+    "create_draft_workspace",
+    "ensure_workspace_id",
+    "get_draft_workspace",
+    "patch_draft_workspace",
+    "summarize_draft_workspace",
+]

src/wf_artifacts/draft_workspaces/api.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import time
+from typing import Any
+
+from wf_artifacts.drafts import patch_workflow_draft, validate_workflow_draft
+
+from .models import WorkflowDraftWorkspace, summarize_draft_workspace
+from .store import DraftWorkspaceConflictError, DraftWorkspaceStore
+
+JsonObject = dict[str, Any]
+JsonPatch = list[dict[str, Any]]
+
+
+def create_draft_workspace(
+    store: DraftWorkspaceStore,
+    *,
+    workspace_id: str,
+    draft: JsonObject,
+    title: str | None = None,
+) -> JsonObject:
+    """Validate and save a new mutable draft workspace."""
+    now = _now_ms()
+    validation = validate_workflow_draft(draft)
+    workspace = WorkflowDraftWorkspace(
+        id=workspace_id,
+        revision=1,
+        title=title,
+        draft=draft,
+        status=validation["status"],
+        diagnostics=validation["diagnostics"],
+        created_at_epoch_ms=now,
+        updated_at_epoch_ms=now,
+    )
+    try:
+        store.create_workspace(workspace)
+    except DraftWorkspaceConflictError as exc:
+        return _conflict_payload(
+            exc.workspace,
+            code="workspace_exists",
+            message=f"draft workspace {workspace_id!r} already exists",
+        )
+    return summarize_draft_workspace(workspace)
+
+
+def patch_draft_workspace(
+    store: DraftWorkspaceStore,
+    *,
+    workspace_id: str,
+    revision: int,
+    patch: JsonPatch,
+) -> JsonObject:
+    """Apply JSON Patch to a stored workspace when the revision matches."""
+    workspace = store.get_workspace(workspace_id)
+    if workspace.revision != revision:
+        return _revision_conflict_payload(workspace, revision)
+    patched = patch_workflow_draft(workspace.draft, patch)
+    if "draft" not in patched:
+        # A malformed JSON Patch is not a draft revision. Return diagnostics
+        # without mutating the stored workspace or burning a revision number.
+        return {
+            **summarize_draft_workspace(workspace),
+            "status": patched["status"],
+            "diagnostics": patched["diagnostics"],
+        }
+    next_workspace = workspace.model_copy(
+        update={
+            "revision": workspace.revision + 1,
+            "draft": patched["draft"],
+            "status": patched["status"],
+            "diagnostics": patched["diagnostics"],
+            "updated_at_epoch_ms": _now_ms(),
+        }
+    )
+    try:
+        store.replace_workspace(next_workspace, expected_revision=revision)
+    except DraftWorkspaceConflictError as exc:
+        return _revision_conflict_payload(exc.workspace, revision)
+    return summarize_draft_workspace(next_workspace)
+
+
+def get_draft_workspace(
+    store: DraftWorkspaceStore,
+    *,
+    workspace_id: str,
+    include_draft: bool = False,
+) -> JsonObject:
+    """Return one stored workspace, compact by default."""
+    return summarize_draft_workspace(
+        store.get_workspace(workspace_id),
+        include_draft=include_draft,
+    )
+
+
+def _now_ms() -> int:
+    return int(time.time() * 1000)
+
+
+def _revision_conflict_payload(
+    workspace: WorkflowDraftWorkspace,
+    expected_revision: int,
+) -> JsonObject:
+    return _conflict_payload(
+        workspace,
+        code="revision_conflict",
+        message=(
+            f"workspace {workspace.id!r} is at revision "
+            f"{workspace.revision}, not {expected_revision}"
+        ),
+    )
+
+
+def _conflict_payload(
+    workspace: WorkflowDraftWorkspace,
+    *,
+    code: str,
+    message: str,
+) -> JsonObject:
+    return {
+        "workspace_id": workspace.id,
+        "revision": workspace.revision,
+        "title": workspace.title,
+        "status": "conflict",
+        "diagnostics": [
+            {
+                "code": code,
+                "path": "revision" if code == "revision_conflict" else "workspace_id",
+                "message": message,
+            }
+        ],
+        "summary": summarize_draft_workspace(workspace)["summary"],
+    }

src/wf_artifacts/draft_workspaces/models.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import re
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+JsonObject = dict[str, Any]
+WORKSPACE_ID_PATTERN = r"^[A-Za-z0-9_.-]+$"
+
+
+def ensure_workspace_id(workspace_id: str) -> str:
+    """Reject ids that cannot be safely used as one local filename stem."""
+    if not re.fullmatch(WORKSPACE_ID_PATTERN, workspace_id):
+        raise ValueError(
+            "workspace_id must match [A-Za-z0-9_.-]+; path separators are not allowed"
+        )
+    return workspace_id
+
+
+class WorkflowDraftWorkspace(BaseModel):
+    """Mutable, revisioned authoring workspace for one workflow draft."""
+
+    id: str = Field(pattern=WORKSPACE_ID_PATTERN)
+    revision: int = Field(default=1, ge=1)
+    title: str | None = None
+    draft: JsonObject
+    status: Literal["valid", "invalid"]
+    diagnostics: list[JsonObject] = Field(default_factory=list)
+    created_at_epoch_ms: int
+    updated_at_epoch_ms: int
+
+
+def summarize_draft_workspace(
+    workspace: WorkflowDraftWorkspace,
+    *,
+    include_draft: bool = False,
+) -> JsonObject:
+    """Return the compact workspace payload used by MCP-facing tools."""
+    steps = workspace.draft.get("steps", {})
+    routes = workspace.draft.get("routes", {})
+    summary: JsonObject = {
+        "workspace_id": workspace.id,
+        "revision": workspace.revision,
+        "title": workspace.title,
+        "status": workspace.status,
+        "diagnostics": workspace.diagnostics,
+        "summary": {
+            "name": workspace.draft.get("name"),
+            "start": workspace.draft.get("start"),
+            "step_count": len(steps) if isinstance(steps, dict) else 0,
+            "route_count": len(routes) if isinstance(routes, dict) else 0,
+            "steps": sorted(steps) if isinstance(steps, dict) else [],
+        },
+    }
+    if include_draft:
+        summary["draft"] = workspace.draft
+    return summary

src/wf_artifacts/draft_workspaces/store.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from threading import RLock
+
+from .models import WorkflowDraftWorkspace, ensure_workspace_id
+
+
+class DraftWorkspaceConflictError(RuntimeError):
+    """Raised when a workspace create/update loses an optimistic-concurrency race."""
+
+    def __init__(self, workspace: WorkflowDraftWorkspace) -> None:
+        self.workspace = workspace
+        super().__init__(
+            f"draft workspace {workspace.id!r} is at revision {workspace.revision}"
+        )
+
+
+class DraftWorkspaceStore:
+    """Storage boundary for mutable workflow draft workspaces."""
+
+    def create_workspace(self, workspace: WorkflowDraftWorkspace) -> None:
+        """Save a new workspace, rejecting duplicate ids."""
+        try:
+            existing = self.get_workspace(workspace.id)
+        except KeyError:
+            self.save_workspace(workspace)
+            return
+        raise DraftWorkspaceConflictError(existing)
+
+    def replace_workspace(
+        self,
+        workspace: WorkflowDraftWorkspace,
+        *,
+        expected_revision: int,
+    ) -> None:
+        """Replace a workspace only if the stored revision still matches."""
+        current = self.get_workspace(workspace.id)
+        if current.revision != expected_revision:
+            raise DraftWorkspaceConflictError(current)
+        self.save_workspace(workspace)
+
+    def save_workspace(self, workspace: WorkflowDraftWorkspace) -> None:
+        raise NotImplementedError
+
+    def get_workspace(self, workspace_id: str) -> WorkflowDraftWorkspace:
+        raise NotImplementedError
+
+    def list_workspaces(self) -> list[WorkflowDraftWorkspace]:
+        raise NotImplementedError
+
+
+class FileDraftWorkspaceStore(DraftWorkspaceStore):
+    """JSON file-backed draft workspace store for local development and tests.
+
+    The internal lock protects optimistic writes inside one broker process. If
+    multiple broker processes share a root, use a stronger store implementation.
+    """
+
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self._lock = RLock()
+        self.workspaces_dir.mkdir(parents=True, exist_ok=True)
+
+    @property
+    def workspaces_dir(self) -> Path:
+        return self.root / "draft_workspaces"
+
+    def save_workspace(self, workspace: WorkflowDraftWorkspace) -> None:
+        with self._lock:
+            self._write_workspace(workspace)
+
+    def create_workspace(self, workspace: WorkflowDraftWorkspace) -> None:
+        with self._lock:
+            path = self._workspace_path(workspace.id)
+            if path.exists():
+                raise DraftWorkspaceConflictError(self.get_workspace(workspace.id))
+            self._write_workspace(workspace)
+
+    def replace_workspace(
+        self,
+        workspace: WorkflowDraftWorkspace,
+        *,
+        expected_revision: int,
+    ) -> None:
+        with self._lock:
+            current = self.get_workspace(workspace.id)
+            if current.revision != expected_revision:
+                raise DraftWorkspaceConflictError(current)
+            self._write_workspace(workspace)
+
+    def _write_workspace(self, workspace: WorkflowDraftWorkspace) -> None:
+        path = self._workspace_path(workspace.id)
+        temp_path = path.with_suffix(".json.tmp")
+        temp_path.write_text(
+            json.dumps(workspace.model_dump(mode="json"), indent=2),
+            encoding="utf-8",
+        )
+        temp_path.replace(path)
+
+    def get_workspace(self, workspace_id: str) -> WorkflowDraftWorkspace:
+        path = self._workspace_path(workspace_id)
+        if not path.exists():
+            raise KeyError(f"unknown draft workspace {workspace_id!r}")
+        return WorkflowDraftWorkspace.model_validate_json(
+            path.read_text(encoding="utf-8")
+        )
+
+    def list_workspaces(self) -> list[WorkflowDraftWorkspace]:
+        return [
+            WorkflowDraftWorkspace.model_validate_json(path.read_text(encoding="utf-8"))
+            for path in sorted(self.workspaces_dir.glob("*.json"))
+        ]
+
+    def _workspace_path(self, workspace_id: str) -> Path:
+        safe_id = ensure_workspace_id(workspace_id)
+        root = self.workspaces_dir.resolve()
+        path = (self.workspaces_dir / f"{safe_id}.json").resolve()
+        if path.parent != root:
+            raise ValueError(f"workspace id escapes workspace store: {workspace_id!r}")
+        return path

src/wf_mcp/broker/config.py

@@ -3,11 +3,12 @@ from __future__ import annotations
 import json
 from pathlib import Path
 
+from wf_artifacts import FileDraftWorkspaceStore, FileWorkflowArtifactStore
+
 from ..control import BrokerConfigFile
 from ..models import BrokerConfig
 from ..sdk import McpSdkAdapter
 from ..storage import FileStore
-from wf_artifacts import FileWorkflowArtifactStore
 from .service import WfMcpService
 
 
@@ -23,6 +24,7 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
     service = WfMcpService(
         store=FileStore(config.store_root),
         artifact_store=FileWorkflowArtifactStore(config.store_root),
+        draft_workspace_store=FileDraftWorkspaceStore(config.store_root),
     )
     for connection in config.connections:
         service.register_connection(connection)

src/wf_mcp/broker/service/core.py

@@ -8,6 +8,8 @@ from typing import Any
 from pydantic import BaseModel
 
 from wf_artifacts import (
+    DraftWorkspaceStore,
+    FileDraftWorkspaceStore,
     FileWorkflowArtifactStore,
     WorkflowArtifact,
     WorkflowArtifactCatalogEntry,
@@ -67,11 +69,16 @@ class WfMcpService:
     event_bus: EventBus = field(default_factory=EventBus)
     include_builtin_specs: bool = True
     artifact_store: WorkflowArtifactStore | None = None
+    draft_workspace_store: DraftWorkspaceStore | None = None
 
     def __post_init__(self) -> None:
         """Install broker-local system specs when enabled."""
         if self.artifact_store is None:
             self.artifact_store = FileWorkflowArtifactStore(_store_root(self.store))
+        if self.draft_workspace_store is None:
+            self.draft_workspace_store = FileDraftWorkspaceStore(
+                _store_root(self.store)
+            )
         if self.include_builtin_specs:
             for source in builtin_sources(self).values():
                 self.register_capability_source(source)

src/wf_mcp/transparent_proxy/runtime.py

@@ -41,6 +41,11 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.create_artifact_from_plan",
     "wf.workflow.create_artifact_from_draft",
     "wf.workflow.patch_draft",
+    "wf.workflow.create_draft_workspace",
+    "wf.workflow.get_draft_workspace",
+    "wf.workflow.patch_draft_workspace",
+    "wf.workflow.create_minimal_draft_workspace",
+    "wf.workflow.create_artifact_from_workspace",
     "wf.workflow.call_capability",
     "wf.workflow.inspect_artifact",
     "wf.workflow.list_deployments",

src/wf_mcp/workflow_surface/handlers.py

@@ -9,12 +9,16 @@ from wf_artifacts import (
     AvailableSource,
     DependencyDiagnostic,
     DiagnosticSeverity,
+    DraftWorkspaceStore,
     RequiredCapability,
     WorkflowArtifact,
     WorkflowCapabilityRef,
     WorkflowDeployment,
     compile_workflow_draft,
+    create_draft_workspace as create_draft_workspace_record,
     create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
+    get_draft_workspace as get_draft_workspace_record,
+    patch_draft_workspace as patch_draft_workspace_record,
     patch_workflow_draft,
     validate_workflow_draft,
     validate_deployment_dependencies,
@@ -340,7 +344,7 @@ class WorkflowSurfaceHandlers:
         required_sources = sorted(
             {
                 capability.logical_source
-                for capability in workflow_artifact.required_capabilities.values()
+                for capability in dict(workflow_artifact.required_capabilities).values()
             }
         )
         return {
@@ -359,6 +363,139 @@ class WorkflowSurfaceHandlers:
     ) -> dict[str, Any]:
         return patch_workflow_draft(draft, patch)
 
+    def _draft_store(self) -> DraftWorkspaceStore:
+        if self.service.draft_workspace_store is None:
+            raise KeyError("draft workspace store is not configured")
+        return self.service.draft_workspace_store
+
+    async def create_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        draft: dict[str, Any],
+        title: str | None = None,
+    ) -> dict[str, Any]:
+        return create_draft_workspace_record(
+            self._draft_store(),
+            workspace_id=workspace_id,
+            draft=draft,
+            title=title,
+        )
+
+    async def get_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        include_draft: bool = False,
+    ) -> dict[str, Any]:
+        return get_draft_workspace_record(
+            self._draft_store(),
+            workspace_id=workspace_id,
+            include_draft=include_draft,
+        )
+
+    async def patch_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        patch: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        return patch_draft_workspace_record(
+            self._draft_store(),
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
+    async def create_minimal_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        name: str,
+        capability_name: str,
+        input_schema: dict[str, Any],
+        state_schema: dict[str, Any],
+        output_schema: dict[str, Any],
+        input_map: dict[str, str],
+        output_map: dict[str, str],
+        error_message_source: str | None = None,
+        title: str | None = None,
+    ) -> dict[str, Any]:
+        """Bootstrap the smallest patchable draft around one workflow capability."""
+        outcomes = self._outcomes_for_capability(capability_name) or ("ok",)
+        steps: dict[str, Any] = {
+            "call": {
+                "use": capability_name,
+                "in": input_map,
+                "out": output_map,
+            }
+        }
+        routes: dict[str, dict[str, str]] = {"call": {"ok": "__end__"}}
+        error_source = error_message_source or _first_state_path(output_map)
+        if "error" in outcomes and error_source is not None:
+            # The bootstrapper cannot infer provider-specific error envelopes.
+            # It only wires an error route when the caller gave, or output_map
+            # exposes, a concrete state path that can become a runtime message.
+            steps["tool_error"] = {
+                "use": "wf.std.runtime_error",
+                "in": {error_source: "message"},
+                "out": {},
+            }
+            routes["call"]["error"] = "tool_error"
+            routes["tool_error"] = {"ok": "__end__"}
+        draft = {
+            "name": name,
+            "input_schema": input_schema,
+            "state_schema": state_schema,
+            "output_schema": output_schema,
+            "start": "call",
+            "steps": steps,
+            "routes": routes,
+        }
+        return await self.create_draft_workspace(
+            workspace_id=workspace_id,
+            title=title,
+            draft=draft,
+        )
+
+    async def create_artifact_from_workspace(
+        self,
+        *,
+        workspace_id: str,
+        artifact_id: str,
+        version: int,
+        title: str,
+        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]:
+        workspace = self._draft_store().get_workspace(workspace_id)
+        validation = await self.validate_draft(draft=workspace.draft)
+        if validation["status"] != "valid":
+            return {
+                "saved": False,
+                "workspace_id": workspace_id,
+                "revision": workspace.revision,
+                "status": validation["status"],
+                "diagnostics": validation["diagnostics"],
+            }
+        return await self.create_artifact_from_draft(
+            artifact_id=artifact_id,
+            version=version,
+            title=title,
+            kind=kind,
+            description=description,
+            draft=workspace.draft,
+            outcomes=outcomes,
+            required_capabilities=required_capabilities,
+            source_bindings=source_bindings,
+            created_from_catalog_version=created_from_catalog_version,
+        )
+
     def _outcomes_for_capability(self, qualified_name: str) -> tuple[str, ...] | None:
         try:
             return self.service._get_qualified_spec(qualified_name).outcomes
@@ -583,6 +720,14 @@ def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     return sorted(str(name) for name in properties)
 
 
+def _first_state_path(output_map: dict[str, str]) -> str | None:
+    """Return the first mapped state path for minimal error-route bootstraps."""
+    for target in output_map.values():
+        if target.startswith("state."):
+            return target
+    return None
+
+
 def _source_id_for_capability(
     sources: dict[str, CapabilitySource],
     qualified_name: str,

src/wf_mcp/workflow_surface/models.py

@@ -32,3 +32,23 @@ class CallCapabilityResult(BaseModel):
         default_factory=list,
         description="Structured diagnostics. Empty for successful calls.",
     )
+
+
+class DraftWorkspaceResult(BaseModel):
+    """Inspector-visible response contract for draft workspace operations."""
+
+    workspace_id: str = Field(description="Mutable draft workspace id.")
+    revision: int = Field(description="Current optimistic-concurrency revision.")
+    title: str | None = Field(default=None, description="Optional workspace title.")
+    status: str = Field(description="Workspace validation status.")
+    diagnostics: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Structured diagnostics for invalid or conflicted workspaces.",
+    )
+    summary: dict[str, Any] = Field(
+        description="Compact draft summary for progressive MCP clients."
+    )
+    draft: dict[str, Any] | None = Field(
+        default=None,
+        description="Full draft document, only returned when requested.",
+    )

src/wf_mcp/workflow_surface/tools.py

@@ -9,7 +9,7 @@ from wf_artifacts.models import RequiredCapability
 from wf_mcp.broker.service import WfMcpService
 
 from .handlers import WorkflowSurfaceHandlers
-from .models import CallCapabilityResult
+from .models import CallCapabilityResult, DraftWorkspaceResult
 
 
 def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None:
@@ -190,6 +190,136 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
     ) -> dict[str, Any]:
         return await handlers.patch_draft(draft=draft, patch=patch)
 
+    @server.tool(
+        name="wf.workflow.create_draft_workspace",
+        title="Create Draft Workspace",
+        description="Store a mutable workflow draft workspace for iterative patching.",
+    )
+    async def create_draft_workspace(
+        workspace_id: str,
+        draft: dict[str, Any],
+        title: str | None = None,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.create_draft_workspace(
+                workspace_id=workspace_id,
+                draft=draft,
+                title=title,
+            )
+        )
+
+    @server.tool(
+        name="wf.workflow.get_draft_workspace",
+        title="Get Draft Workspace",
+        description="Fetch a mutable workflow draft workspace by id.",
+    )
+    async def get_draft_workspace(
+        workspace_id: str,
+        include_draft: bool = False,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.get_draft_workspace(
+                workspace_id=workspace_id,
+                include_draft=include_draft,
+            )
+        )
+
+    @server.tool(
+        name="wf.workflow.patch_draft_workspace",
+        title="Patch Draft Workspace",
+        description=(
+            "Apply an RFC 6902 JSON Patch to a stored workflow draft workspace "
+            "when the expected revision matches."
+        ),
+    )
+    async def patch_draft_workspace(
+        workspace_id: str,
+        revision: int,
+        patch: list[dict[str, Any]],
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.patch_draft_workspace(
+                workspace_id=workspace_id,
+                revision=revision,
+                patch=patch,
+            )
+        )
+
+    @server.tool(
+        name="wf.workflow.create_minimal_draft_workspace",
+        title="Create Minimal Draft Workspace",
+        description="Bootstrap a patchable draft workspace around one capability.",
+    )
+    async def create_minimal_draft_workspace(
+        workspace_id: str,
+        name: str,
+        capability_name: str,
+        input_schema: dict[str, Any],
+        state_schema: dict[str, Any],
+        output_schema: dict[str, Any],
+        input_map: dict[str, str],
+        output_map: dict[str, str],
+        error_message_source: str | None = None,
+        title: str | None = None,
+    ) -> DraftWorkspaceResult:
+        return DraftWorkspaceResult.model_validate(
+            await handlers.create_minimal_draft_workspace(
+                workspace_id=workspace_id,
+                name=name,
+                capability_name=capability_name,
+                input_schema=input_schema,
+                state_schema=state_schema,
+                output_schema=output_schema,
+                input_map=input_map,
+                output_map=output_map,
+                error_message_source=error_message_source,
+                title=title,
+            )
+        )
+
+    @server.tool(
+        name="wf.workflow.create_artifact_from_workspace",
+        title="Create Workflow Artifact From Workspace",
+        description=(
+            "Validate the current draft workspace and save it as a versioned "
+            "workflow artifact."
+        ),
+    )
+    async def create_artifact_from_workspace(
+        workspace_id: str,
+        artifact_id: str,
+        version: int,
+        title: str,
+        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_workspace(
+            workspace_id=workspace_id,
+            artifact_id=artifact_id,
+            version=version,
+            title=title,
+            kind=kind,
+            description=description,
+            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.inspect_artifact",
         title="Inspect Workflow Artifact",

tests/artifacts/test_draft_workspaces.py

@@ -0,0 +1,263 @@
+from __future__ import annotations
+
+from typing import Any
+
+from wf_artifacts import (
+    FileDraftWorkspaceStore,
+    WorkflowDraftWorkspace,
+    create_draft_workspace,
+    get_draft_workspace,
+    patch_draft_workspace,
+    summarize_draft_workspace,
+)
+
+
+def test_draft_workspace_stores_mutable_draft_with_revision() -> None:
+    workspace = WorkflowDraftWorkspace(
+        id="echo_draft",
+        revision=1,
+        title="Echo Draft",
+        draft=_draft(),
+        status="valid",
+        diagnostics=[],
+        created_at_epoch_ms=100,
+        updated_at_epoch_ms=100,
+    )
+
+    assert workspace.id == "echo_draft"
+    assert workspace.revision == 1
+    assert workspace.draft["steps"]["echo"]["use"] == "demo.echo"
+
+
+def test_draft_workspace_summary_is_compact() -> None:
+    workspace = WorkflowDraftWorkspace(
+        id="echo_draft",
+        revision=3,
+        draft=_draft(),
+        status="valid",
+        diagnostics=[],
+        created_at_epoch_ms=100,
+        updated_at_epoch_ms=200,
+    )
+
+    summary = summarize_draft_workspace(workspace)
+
+    assert summary["workspace_id"] == "echo_draft"
+    assert summary["revision"] == 3
+    assert summary["status"] == "valid"
+    assert summary["summary"]["name"] == "echo"
+    assert summary["summary"]["steps"] == ["echo"]
+    assert "draft" not in summary
+
+
+def test_file_draft_workspace_store_round_trips_workspace(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    workspace = WorkflowDraftWorkspace(
+        id="echo_draft",
+        revision=1,
+        draft=_draft(),
+        status="valid",
+        diagnostics=[],
+        created_at_epoch_ms=100,
+        updated_at_epoch_ms=100,
+    )
+
+    store.save_workspace(workspace)
+    loaded = store.get_workspace("echo_draft")
+
+    assert loaded == workspace
+
+
+def test_file_draft_workspace_store_lists_workspaces(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    store.save_workspace(
+        WorkflowDraftWorkspace(
+            id="b",
+            revision=1,
+            draft=_draft(),
+            status="valid",
+            diagnostics=[],
+            created_at_epoch_ms=100,
+            updated_at_epoch_ms=100,
+        )
+    )
+    store.save_workspace(
+        WorkflowDraftWorkspace(
+            id="a",
+            revision=1,
+            draft=_draft(),
+            status="valid",
+            diagnostics=[],
+            created_at_epoch_ms=100,
+            updated_at_epoch_ms=100,
+        )
+    )
+
+    assert [workspace.id for workspace in store.list_workspaces()] == ["a", "b"]
+
+
+def test_file_draft_workspace_store_rejects_path_traversal_ids(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+
+    try:
+        store.get_workspace("../outside")
+    except ValueError as exc:
+        assert "path separators are not allowed" in str(exc)
+    else:
+        raise AssertionError("expected unsafe workspace id to be rejected")
+
+
+def test_create_draft_workspace_validates_and_saves(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+
+    result = create_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        draft=_draft(),
+        title="Echo Draft",
+    )
+
+    loaded = store.get_workspace("echo_draft")
+    assert result["workspace_id"] == "echo_draft"
+    assert result["status"] == "valid"
+    assert loaded.revision == 1
+    assert loaded.status == "valid"
+
+
+def test_create_draft_workspace_rejects_duplicate_id(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    create_draft_workspace(store, workspace_id="echo_draft", draft=_draft())
+
+    result = create_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        draft=_draft(),
+        title="Replacement",
+    )
+
+    loaded = store.get_workspace("echo_draft")
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "workspace_exists"
+    assert loaded.revision == 1
+    assert loaded.title is None
+
+
+def test_patch_draft_workspace_applies_patch_and_increments_revision(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    create_draft_workspace(store, workspace_id="echo_draft", draft=_draft())
+
+    result = patch_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        revision=1,
+        patch=[
+            {
+                "op": "replace",
+                "path": "/name",
+                "value": "echo_v2",
+            }
+        ],
+    )
+
+    loaded = store.get_workspace("echo_draft")
+    assert result["revision"] == 2
+    assert result["status"] == "valid"
+    assert loaded.revision == 2
+    assert loaded.draft["name"] == "echo_v2"
+
+
+def test_patch_draft_workspace_rejects_stale_revision(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    create_draft_workspace(store, workspace_id="echo_draft", draft=_draft())
+    patch_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        revision=1,
+        patch=[],
+    )
+
+    result = patch_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        revision=1,
+        patch=[],
+    )
+
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+    assert store.get_workspace("echo_draft").revision == 2
+
+
+def test_patch_draft_workspace_rejects_duplicate_revision_patch(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    create_draft_workspace(store, workspace_id="echo_draft", draft=_draft())
+
+    first = patch_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        revision=1,
+        patch=[{"op": "replace", "path": "/name", "value": "first"}],
+    )
+    second = patch_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        revision=1,
+        patch=[{"op": "replace", "path": "/name", "value": "second"}],
+    )
+
+    loaded = store.get_workspace("echo_draft")
+    assert first["revision"] == 2
+    assert second["status"] == "conflict"
+    assert second["diagnostics"][0]["code"] == "revision_conflict"
+    assert loaded.draft["name"] == "first"
+
+
+def test_patch_draft_workspace_rejects_invalid_patch_without_revision_bump(
+    tmp_path,
+) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    create_draft_workspace(store, workspace_id="echo_draft", draft=_draft())
+
+    result = patch_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        revision=1,
+        patch=[{"op": "remove", "path": "/missing"}],
+    )
+
+    assert result["status"] == "invalid"
+    assert result["diagnostics"][0]["code"] == "patch_invalid"
+    assert store.get_workspace("echo_draft").revision == 1
+
+
+def test_get_draft_workspace_includes_full_draft_only_when_requested(tmp_path) -> None:
+    store = FileDraftWorkspaceStore(tmp_path)
+    create_draft_workspace(store, workspace_id="echo_draft", draft=_draft())
+
+    compact = get_draft_workspace(store, workspace_id="echo_draft")
+    full = get_draft_workspace(
+        store,
+        workspace_id="echo_draft",
+        include_draft=True,
+    )
+
+    assert "draft" not in compact
+    assert full["draft"]["steps"]["echo"]["use"] == "demo.echo"
+
+
+def _draft() -> dict[str, Any]:
+    return {
+        "name": "echo",
+        "input_schema": {"type": "object", "properties": {}},
+        "state_schema": {"fields": {"echoed": {"type": "string"}}},
+        "output_schema": {"type": "object", "properties": {}},
+        "start": "echo",
+        "steps": {
+            "echo": {
+                "use": "demo.echo",
+                "in": {},
+                "out": {"echoed": "state.echoed"},
+            }
+        },
+        "routes": {"echo": {"ok": "__end__"}},
+    }

tests/wf_mcp/test_server.py

@@ -64,6 +64,11 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.compile_draft" in names
             assert "wf.workflow.create_artifact_from_draft" in names
             assert "wf.workflow.patch_draft" in names
+            assert "wf.workflow.create_draft_workspace" in names
+            assert "wf.workflow.get_draft_workspace" in names
+            assert "wf.workflow.patch_draft_workspace" in names
+            assert "wf.workflow.create_minimal_draft_workspace" in names
+            assert "wf.workflow.create_artifact_from_workspace" in names
             assert "wf.workflow.run_deployment" in names
             call_capability_schema = tools_by_name[
                 "wf.workflow.call_capability"
@@ -72,6 +77,12 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "source_id" in call_capability_schema["properties"]
             assert "kind" in call_capability_schema["properties"]
             assert "diagnostics" in call_capability_schema["properties"]
+            create_workspace_schema = tools_by_name[
+                "wf.workflow.create_draft_workspace"
+            ].outputSchema
+            assert create_workspace_schema is not None
+            assert "workspace_id" in create_workspace_schema["properties"]
+            assert "revision" in create_workspace_schema["properties"]
 
             echo_result = await client.call_tool(
                 "fixture.personal.echo_tool",
@@ -165,6 +176,11 @@ def test_server_search_mode_pins_stable_control_and_workflow_tools() -> None:
             assert "wf.workflow.list_artifacts" in names
             assert "wf.workflow.validate_draft" in names
             assert "wf.workflow.create_artifact_from_draft" in names
+            assert "wf.workflow.create_draft_workspace" in names
+            assert "wf.workflow.get_draft_workspace" in names
+            assert "wf.workflow.patch_draft_workspace" in names
+            assert "wf.workflow.create_minimal_draft_workspace" in names
+            assert "wf.workflow.create_artifact_from_workspace" in names
             assert "wf.workflow.call_capability" in names
             assert "wf.workflow.inspect_artifact" in names
             assert "wf.workflow.list_deployments" in names

tests/wf_mcp/test_service.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import shutil
 
+from wf_artifacts import FileDraftWorkspaceStore
 from wf_authoring import NodeSpec
 from wf_core import END, NodeUse, RunStatus
 from wf_mcp.broker import WfMcpService
@@ -103,6 +104,14 @@ def test_service_installs_builtin_stdlib_specs_by_default() -> None:
     )
 
 
+def test_service_installs_default_draft_workspace_store() -> None:
+    root = local_temp_root() / "service_default_draft_workspace_store"
+    service = WfMcpService(store=FileStore(root))
+
+    assert isinstance(service.draft_workspace_store, FileDraftWorkspaceStore)
+    assert service.draft_workspace_store.root == root
+
+
 def test_service_registers_empty_source_for_connection_without_catalog() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "empty_source"))
 

tests/wf_mcp/test_workflow_surface.py

@@ -52,6 +52,12 @@ def changed_echo_tool(payload: ChangedEchoInput) -> ChangedEchoOutput:
     return ChangedEchoOutput(echoed=payload.message)
 
 
+@node(name="mcp_echo_tool", outcomes=("ok", "error"))
+def mcp_echo_tool(payload: ChangedEchoInput) -> ChangedEchoOutput:
+    """Test fixture that mirrors naive MCP wrappers with ok/error outcomes."""
+    return ChangedEchoOutput(echoed=payload.message)
+
+
 @reducer(name="custom.default.multiply")
 def multiply(current: int | None, incoming: int) -> int:
     return (current or 1) * incoming
@@ -348,6 +354,133 @@ def test_workflow_surface_patches_draft_without_saving() -> None:
     assert not artifact_store.list_artifacts()
 
 
+def test_workflow_surface_creates_and_gets_draft_workspace() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "surface_workspace")
+    handlers = _handlers(artifact_store)
+
+    created = asyncio.run(
+        handlers.create_draft_workspace(
+            workspace_id="echo_draft",
+            title="Echo Draft",
+            draft=_echo_draft(),
+        )
+    )
+    fetched = asyncio.run(
+        handlers.get_draft_workspace(
+            workspace_id="echo_draft",
+            include_draft=True,
+        )
+    )
+
+    assert created["workspace_id"] == "echo_draft"
+    assert created["revision"] == 1
+    assert fetched["draft"]["steps"]["echo"]["use"] == "demo.personal.echo_tool"
+
+
+def test_workflow_surface_patches_draft_workspace_by_revision() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_workspace_patch"
+    )
+    handlers = _handlers(artifact_store)
+    asyncio.run(
+        handlers.create_draft_workspace(
+            workspace_id="echo_draft",
+            draft=_echo_draft(),
+        )
+    )
+
+    patched = asyncio.run(
+        handlers.patch_draft_workspace(
+            workspace_id="echo_draft",
+            revision=1,
+            patch=[{"op": "replace", "path": "/name", "value": "echo_v2"}],
+        )
+    )
+
+    assert patched["revision"] == 2
+    assert patched["status"] == "valid"
+
+
+def test_workflow_surface_creates_minimal_draft_workspace_with_error_route() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_minimal_workspace"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_minimal_workspace_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", mcp_echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+
+    result = asyncio.run(
+        handlers.create_minimal_draft_workspace(
+            workspace_id="echo_draft",
+            name="echo",
+            capability_name="demo.personal.mcp_echo_tool",
+            input_schema={
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
+            },
+            state_schema={"fields": {"echoed": {"type": "string"}}},
+            output_schema={
+                "type": "object",
+                "properties": {"echoed": {"type": "string"}},
+                "required": ["echoed"],
+            },
+            input_map={"input.text": "text"},
+            output_map={"echoed": "state.echoed"},
+        )
+    )
+    assert service.draft_workspace_store is not None
+    workspace = service.draft_workspace_store.get_workspace("echo_draft")
+
+    assert result["workspace_id"] == "echo_draft"
+    assert workspace.draft["routes"]["call"]["ok"] == "__end__"
+    assert workspace.draft["routes"]["call"]["error"] == "tool_error"
+    assert workspace.draft["steps"]["tool_error"]["use"] == "wf.std.runtime_error"
+    assert workspace.draft["steps"]["tool_error"]["in"] == {"state.echoed": "message"}
+
+
+def test_workflow_surface_creates_artifact_from_workspace() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_workspace_artifact"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_workspace_artifact_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+    handlers = WorkflowSurfaceHandlers(service)
+    asyncio.run(
+        handlers.create_draft_workspace(
+            workspace_id="echo_draft",
+            draft=_echo_draft(),
+        )
+    )
+
+    result = asyncio.run(
+        handlers.create_artifact_from_workspace(
+            workspace_id="echo_draft",
+            artifact_id="workspace_echo",
+            version=1,
+            title="Workspace Echo",
+            outcomes=("completed",),
+            source_bindings={"demo": "demo.personal"},
+        )
+    )
+
+    artifact = artifact_store.get_artifact("workspace_echo", 1)
+    assert result["saved"] is True
+    assert artifact.id == "workspace_echo"
+
+
 def test_raw_workflow_plan_uses_core_step_and_edge_models() -> None:
     plan = RawWorkflowPlan.model_validate(_echo_artifact().plan)
 
@@ -696,7 +829,7 @@ def test_workflow_surface_calls_saved_wrapper_artifact_with_deployment_bindings(
 
 def _handlers(artifact_store: FileWorkflowArtifactStore) -> WorkflowSurfaceHandlers:
     service = WfMcpService(
-        store=FileStore(local_temp_root() / "surface_mcp"),
+        store=FileStore(artifact_store.root / "surface_mcp" / str(id(artifact_store))),
         artifact_store=artifact_store,
     )
     return WorkflowSurfaceHandlers(service)