MCP surface: slim payloads, pinned workspaces, costs

docs/MCP.md

@@ -89,6 +89,8 @@ If `dw.serve` runs on another machine with `--mcp` (see
     claude mcp add --transport http dw http://<box>:8765/mcp \
       --header "Authorization: Bearer <token>"
 
+The same token fetches generated files: see step 7 of `The loop` in [WORKFLOW_GUIDE.md](WORKFLOW_GUIDE.md#the-loop).
+
 Nothing from this repository is installed on the client. The stdio setup
 below is for a machine that has its own `dw` install, and also works
 against a remote `--url` with `--token`.
@@ -247,10 +249,10 @@ The session starts in `default` and stays there unless it is told otherwise.
 
 | Tool | Arguments | Purpose |
 | --- | --- | --- |
-| `validate_workflow(workflow=None, name=None)` | exactly one of `workflow` (inline definition) or `name` (a stored workflow, as `list_workflows` reports it) | Check a workflow against the schema and against real pipeline signatures. Free and instant. Validating by name uses the workflow file's own directory as the base directory, so it sees what a run would. Returns every schema violation in `errors`, each with the JSON path it sits at, so a draft is fixed in one pass |
+| `validate_workflow(workflow=None, name=None, workspace=None)` | exactly one of `workflow` (inline definition) or `name` (a stored workflow, as `list_workflows` reports it), optional `workspace` | Check a workflow against the schema and against real pipeline signatures. Free and instant. Validating by name uses the workflow file's own directory as the base directory, so it sees what a run would. Returns every schema violation in `errors`, each with the JSON path it sits at, so a draft is fixed in one pass. `workspace` names the workspace for this one call without switching the session to it - use it to pin a job whose `output:` or `asset:` references live in a workspace other than the session's |
 | `list_workspaces()` | — | The server's workspaces and which one this session is using. Each has its own workflows, assets and outputs; the prompt library is shared by all of them |
 | `use_workspace(name)` | `name` | Work in that workspace for the rest of the session - every later call reads and writes there. This is how to keep your work out of another agent's namespace rather than sharing the default one. Checked against the server, so a typo fails here rather than scoping every later call to nothing |
-| `create_workspace(name)` | `name` | Create a workspace. Creating does not switch to it |
+| `create_workspace(name, use=False)` | `name`, `use` | Create a workspace. Pass use=true to switch this session to it as well; otherwise the session stays where it was and the result says so |
 | `delete_workspace(name, acknowledged_cost=False)` | `name`, `acknowledged_cost` | Permanently delete a workspace and everything in it. Refuses without the acknowledgement, reporting what it would remove |
 | `list_assets()` | — | The input media on the server, each with the `asset:` reference a workflow argument carries. Look here before asking for a file - what a workflow needs may already be there |
 | `keep_output(name, asset_name=None, overwrite=False)` | `name`, optional `asset_name`, `overwrite` | Keep a generated file as an input asset under a stable `asset:` name, so a later workflow can rely on it. The copy happens on the server: nothing is downloaded or re-uploaded |
@@ -279,12 +281,12 @@ references written in the same session.
 
 | Tool | Arguments | Purpose |
 | --- | --- | --- |
-| `run_workflow(workflow_path=None, inline_workflow=None, arguments=None, acknowledged_cost=False)` | exactly one of `workflow_path` (a catalog name from `list_workflows`, with or without `.json`, or a path to a workflow file on the server) or `inline_workflow`, optional `arguments`, `acknowledged_cost` | Queue a workflow for generation. Returns as soon as the job is queued |
+| `run_workflow(workflow_path=None, inline_workflow=None, arguments=None, acknowledged_cost=False, workspace=None)` | exactly one of `workflow_path` (a catalog name from `list_workflows`, with or without `.json`, or a path to a workflow file on the server) or `inline_workflow`, optional `arguments`, `acknowledged_cost`, `workspace` | Queue a workflow for generation. Returns as soon as the job is queued. `workspace` names the workspace for this one call without switching the session to it - use it to pin a job whose `output:` or `asset:` references live in a workspace other than the session's |
 | `get_job(job_id)` | `job_id` | Get a job's status, warnings, output manifest, error and traceback |
 | `get_job_workflow(job_id)` | `job_id` | The workflow the job actually ran. `realized: true` means every mutable input is pinned (arguments, seed, prompts, `output:latest`); `false` means the job predates run tracking and this is the definition as submitted. Pass it to `save_workflow` to keep it under a name |
-| `export_job(job_id, overwrite=False)` | `job_id`, `overwrite` | Gather one finished job into `<workspace>/exports/<job id>/` on the server: the realized workflow, the run's manifest, the job row, a README, and copies of the assets, earlier-run inputs and outputs. Returns the directory, a zip URL, the file list with sizes and the three JSON files inline. **The directory is on the machine running the server**, like `download_output`'s destination - fetch the zip URL and unpack it into `exports/` under the session's working directory (a deliverable, not a temp file); the archive already unpacks into one folder named after the job id |
+| `export_job(job_id, overwrite=False)` | `job_id`, `overwrite` | Gather one finished job into `<workspace>/exports/<job id>/` on the server: the realized workflow, the run's manifest, the job row, a README, and copies of the assets, earlier-run inputs and outputs. Returns the directory, a zip URL, the file list with sizes and the total. The three JSON files are in the zip, not repeated here - get_job_workflow and get_job serve them individually. **The directory is on the machine running the server**, like `download_output`'s destination - fetch the zip URL and unpack it into `exports/` under the session's working directory (a deliverable, not a temp file); the archive already unpacks into one folder named after the job id |
 | `get_job_events(job_id, after=-1, limit=200)` | `job_id`, `after`, `limit` | Get a page of a job's progress events |
-| `wait_for_job(job_id, timeout_seconds=20)` | `job_id`, `timeout_seconds` | Block until a job reaches a terminal status, or `timeout_seconds` elapses (capped well under a generation's real runtime). Use instead of hand-polling `get_job`/`get_job_events` in a loop; if it returns `still_running: true`, call it again |
+| `wait_for_job(job_id, timeout_seconds=20)` | `job_id`, `timeout_seconds` | Block until a job reaches a terminal status, or `timeout_seconds` elapses (capped well under a generation's real runtime). Use instead of hand-polling `get_job`/`get_job_events` in a loop; if it returns `still_running: true`, call it again. Returns a slim job - status, warnings, error, and the manifest once finished - without the arguments; `get_job` has those |
 | `cancel_job(job_id)` | `job_id` | Ask a queued or running job to stop |
 | `rerun_job(job_id, acknowledged_cost=False, new_seed=False)` | `job_id`, `acknowledged_cost`, `new_seed` | Queue a fresh job from a previous job's stored specification. Costs GPU time, so it passes the same gate as `run_workflow`. `new_seed=true` draws a fresh seed into the workflow's seed variable — without it a seeded workflow's rerun repeats its arguments exactly and the step cache serves the whole run from the earlier one's files (`reused: true`), generating nothing. `get_job_workflow`'s `seed_variable` says whether there is one |
 | `move_job(job_id, direction)` | `job_id`, `direction` (`up`\|`down`\|`front`\|`back`) | Reorder a queued job |

docs/WORKFLOW_GUIDE.md

@@ -343,6 +343,17 @@ signal to restructure the workflow, not to add another reference.
    `still_running: true`.
 6. `get_output_image` to look at what was actually made, and say whether it
    answers the request. Nothing before this step establishes that it does.
+7. Getting the files to the user's machine. `download_output` and `export_job`
+   write on the machine running `dw.serve`, which over a remote `--mcp`
+   endpoint is the GPU box. The last mile of every deliverable is the `url`
+   each `list_gallery` entry carries (or `export_job`'s `zip_url`), fetched
+   with the same bearer token the MCP connection uses:
+
+       curl -H "Authorization: Bearer $DW_API_TOKEN" \
+            -o exports/still.png "http://<box>:8765/outputs/ltx2/Gyre/20260910-.../still.png"
+
+   Put the result under `exports/` in the session's working directory - it is
+   the user's deliverable, not a temporary file.
 
 ### Keeping a set consistent
 

dw/server/catalog_shape.py

@@ -175,8 +175,7 @@ def _cuts_together(steps):
                 sources |= {
                     item
                     for item in videos
-                    if isinstance(item, str)
-                    and not item.startswith("previous_result:")
+                    if isinstance(item, str) and not item.startswith("previous_result:")
                 }
             if len(sources) >= 2:
                 return True

dw_mcp/CLAUDE.md

@@ -17,6 +17,10 @@ means something else to the engine), `use_workspace` to switch, and
 `DwClient._scoped` adds the selector to every request's query string so no
 handler carries a workspace parameter. The default sends nothing, so a
 session that never chooses looks exactly like one from before workspaces.
+`run_workflow` and `validate_workflow` are the two exceptions: each takes an
+optional per-call `workspace` that `_scoped`'s `setdefault` lets win over the
+session's, so one call can be pinned to a workspace other than the session's
+without switching it.
 `get_server_info`
 (`/api/server`) is the capability call: the device a run will use, the dw
 version, the workspace and the workflow/output/prompt/asset directories, which is what tells an

dw_mcp/authoring.py

@@ -9,7 +9,7 @@
 from dw_mcp.client import DwApiError, api_path
 
 
-def validate_workflow(client, workflow=None, name=None):
+def validate_workflow(client, workflow=None, name=None, workspace=None):
     """Schema- and signature-check a workflow without queuing anything. This
     is free (no GPU work) and should be called before any run or save. Give
     either an inline definition or the name of a stored one, as
@@ -19,11 +19,12 @@ def validate_workflow(client, workflow=None, name=None):
             "Provide exactly one of `workflow` (an inline definition) or "
             "`name` (a stored workflow)."
         )
+    params = {"workspace": workspace} if workspace else None
     if workflow is None:
         # The server resolves the name against its own workflow directory,
         # so validation sees the same base directory a run would
-        return client.post_json("/api/validate", {"workflow_path": name})
-    return client.post_json("/api/validate", {"workflow": workflow})
+        return client.post_json("/api/validate", {"workflow_path": name}, params=params)
+    return client.post_json("/api/validate", {"workflow": workflow}, params=params)
 
 
 def save_workflow(client, name, workflow):

dw_mcp/diagnose.py

@@ -36,6 +36,7 @@ def run_workflow(
     inline_workflow=None,
     arguments=None,
     acknowledged_cost=False,
+    workspace=None,
 ):
     """Queue a workflow. `workflow_path` is either a catalog name from
     `list_workflows` or a path to a workflow file on the server. Returns as
@@ -56,7 +57,11 @@ def run_workflow(
         payload["workflow"] = inline_workflow
     # base_dir is deliberately absent: it decides where an inline workflow's
     # relative paths resolve, and the MCP surface does not hand that out
-    job = client.post_json("/api/jobs", payload)
+    # A named workspace pins this one job rather than the session: a
+    # restarted session forgets use_workspace, and a job that resolves
+    # output: references in the wrong root fails after it was queued
+    params = {"workspace": workspace} if workspace else None
+    job = client.post_json("/api/jobs", payload, params=params)
     return {
         "job_id": job.get("id"),
         "status": job.get("status"),
@@ -100,6 +105,35 @@ def get_job_events(client, job_id, after=-1, limit=200):
     )
 
 
+_SLIM_KEYS = (
+    "id",
+    "workflow_name",
+    "status",
+    "created_at",
+    "started_at",
+    "finished_at",
+    "workspace",
+    "run_id",
+    "queue_position",
+    "warnings",
+    "error",
+    "event_count",
+)
+
+
+def slim_job(job):
+    """A job row without its arguments and traceback - what a poll needs.
+
+    The arguments of an H3 workflow are thousands of tokens of prompt text,
+    repeated on every poll of a long render; get_job serves them once. The
+    manifest is kept only once the job is terminal, when it names files.
+    """
+    slim = {key: job.get(key) for key in _SLIM_KEYS if key in job}
+    if job.get("status") in TERMINAL_STATUSES:
+        slim["manifest"] = job.get("manifest")
+    return slim
+
+
 def wait_for_job(client, job_id, timeout_seconds=20):
     """Block until a job reaches a terminal status, or `timeout_seconds`
     elapses - a bounded alternative to polling `get_job`/`get_job_events` by
@@ -111,7 +145,9 @@ def wait_for_job(client, job_id, timeout_seconds=20):
     so this never blocks past a budget kept well under that. Returns as
     soon as the job's status is succeeded, failed or cancelled. If the
     timeout elapses first, returns the job's last-seen status with
-    `still_running: true` instead of hanging - call again to keep waiting."""
+    `still_running: true` instead of hanging - call again to keep waiting.
+    Returns a slim job - status, warnings, error, and the manifest once
+    finished - without the arguments; get_job has those."""
     timeout_seconds = max(0.0, min(float(timeout_seconds), MAX_WAIT_SECONDS))
     deadline = time.monotonic() + timeout_seconds
     while True:
@@ -122,15 +158,17 @@ def wait_for_job(client, job_id, timeout_seconds=20):
                 "job_id": job_id,
                 "status": status,
                 "still_running": False,
-                "job": job,
+                "job": slim_job(job),
+                "next": "get_job(job_id) for the arguments and traceback, "
+                "get_job_workflow(job_id) for the realized workflow.",
             }
         remaining = deadline - time.monotonic()
         if remaining <= 0:
             return {
                 "job_id": job_id,
                 "status": status,
                 "still_running": True,
-                "job": job,
+                "job": slim_job(job),
                 "next": "Call wait_for_job again, or get_job_events for "
                 "incremental progress.",
             }

dw_mcp/exports.py

@@ -16,9 +16,9 @@ def export_job(client, job_id, overwrite=False):
     file rather than linking them, so a video job's export costs its size
     again on the server's disk; `total_bytes` in the result reports what
     was copied. Returns the directory, the zip URL, the file list with
-    sizes and the total, and the three JSON files inline. The directory is
-    on the server machine, not this one - use the zip URL to fetch it
-    elsewhere."""
+    sizes and the total. The three JSON files are in the zip, not repeated
+    here. The directory is on the server machine, not this one - use the
+    zip URL to fetch it elsewhere."""
     body = client.post_json(
         api_path("api", "jobs", job_id, "export"),
         params={"overwrite": "true" if overwrite else "false"},
@@ -32,13 +32,12 @@ def export_job(client, job_id, overwrite=False):
         "files": body.get("files") or [],
         "total_bytes": body.get("total_bytes"),
         "missing": body.get("missing") or [],
-        "workflow": body.get("workflow"),
-        "manifest": body.get("manifest"),
-        "job": body.get("job"),
         "next": "The directory is on the server. To give the user the files, "
         "fetch zip_url and unpack it into exports/ under the session's "
         "working directory - it is the user's deliverable, not a temporary "
         "file, so not a scratch or temp directory. The archive already "
         "unpacks into one folder named after the job id; do not create "
-        "that folder first or the id is doubled in the path.",
+        "that folder first or the id is doubled in the path. workflow.json, "
+        "manifest.json and job.json are inside it - they are not repeated "
+        "here; get_job_workflow and get_job serve them individually.",
     }