workspace draft from a capability

docs/workflow_capabilities.md

@@ -269,6 +269,11 @@ authoring loop:
   - returns one full workflow capability contract with schemas and outcomes
   - includes `wrapper_hints`, a conservative scaffold for creating a wrapper
     draft from the inspected capability
+- `wf.workflow.create_draft_workspace_from_capability`
+  - inspects one workflow capability, applies its `wrapper_hints`, and creates a
+    patchable draft workspace
+  - returns the hints it used so clients can immediately patch uncertain maps,
+    schemas, or routes by revision
 - `wf.workflow.call_capability`
   - executes one such capability once for direct testing
   - returns `qualified_name`, `source_id`, `kind`, optional `deployment_id`,

src/wf_mcp/workflow_surface/handlers.py

@@ -678,6 +678,37 @@ class WorkflowSurfaceHandlers:
             draft=draft,
         )
 
+    async def create_draft_workspace_from_capability(
+        self,
+        *,
+        workspace_id: str,
+        capability_name: str,
+        name: str | None = None,
+        title: str | None = None,
+        input_schema: dict[str, Any] | None = None,
+        state_schema: dict[str, Any] | None = None,
+        output_schema: dict[str, Any] | None = None,
+        input_map: dict[str, str] | None = None,
+        output_map: dict[str, str] | None = None,
+        error_message_source: str | None = None,
+    ) -> dict[str, Any]:
+        """Create a patchable draft workspace from inspect_capability hints."""
+        capability = await self.inspect_capability(qualified_name=capability_name)
+        hints = capability["wrapper_hints"]
+        result = await self.create_minimal_draft_workspace(
+            workspace_id=workspace_id,
+            name=name or _draft_name_from_capability(capability_name),
+            capability_name=capability_name,
+            input_schema=input_schema or hints["input_schema"],
+            state_schema=state_schema or hints["state_schema"],
+            output_schema=output_schema or hints["output_schema"],
+            input_map=input_map or hints["input_map"],
+            output_map=output_map or hints["output_map"],
+            error_message_source=error_message_source,
+            title=title,
+        )
+        return {**result, "wrapper_hints": hints}
+
     async def create_artifact_from_workspace(
         self,
         *,
@@ -993,6 +1024,11 @@ def _escape_json_pointer(value: str) -> str:
     return value.replace("~", "~0").replace("/", "~1")
 
 
+def _draft_name_from_capability(capability_name: str) -> str:
+    """Return a stable draft name when caller does not provide one."""
+    return capability_name.replace(".", "_").replace("-", "_")
+
+
 def _source_id_for_capability(
     sources: dict[str, CapabilitySource],
     qualified_name: str,

src/wf_mcp/workflow_surface/models.py

@@ -233,6 +233,61 @@ class CreateMinimalDraftWorkspaceRequest(BaseModel):
     title: str | None = Field(default=None, description="Optional workspace title.")
 
 
+class CreateDraftWorkspaceFromCapabilityRequest(BaseModel):
+    """Typed MCP request for bootstrapping a draft from wrapper hints."""
+
+    workspace_id: WorkspaceId
+    capability_name: str = Field(
+        description=(
+            "Workflow capability to inspect and call, such as "
+            "demo.default.echo_tool or workflow.echo_wrapper.v1."
+        )
+    )
+    name: str | None = Field(
+        default=None,
+        description="Optional workflow draft name. Defaults to a safe capability name.",
+    )
+    title: str | None = Field(default=None, description="Optional workspace title.")
+    input_schema: JsonSchemaObject | None = Field(
+        default=None,
+        description="Optional override for the hinted public input schema.",
+    )
+    state_schema: JsonSchemaObject | None = Field(
+        default=None,
+        description="Optional override for the hinted workflow state schema.",
+    )
+    output_schema: JsonSchemaObject | None = Field(
+        default=None,
+        description="Optional override for the hinted public output schema.",
+    )
+    input_map: DraftPathMap | None = Field(
+        default=None,
+        description="Optional override for the hinted workflow input map.",
+    )
+    output_map: DraftPathMap | None = Field(
+        default=None,
+        description="Optional override for the hinted capability output map.",
+    )
+    error_message_source: str | None = Field(
+        default=None,
+        description=(
+            "Optional state path used as runtime_error.message when the capability "
+            "has an error outcome, for example state.error_message."
+        ),
+    )
+
+
+class CreateDraftWorkspaceFromCapabilityResult(DraftWorkspaceResult):
+    """Draft workspace result plus the wrapper hints used to bootstrap it."""
+
+    wrapper_hints: dict[str, Any] = Field(
+        description=(
+            "The wrapper_hints payload used before applying request overrides. "
+            "Use this to patch uncertain maps or schemas by revision."
+        )
+    )
+
+
 class CreateArtifactFromWorkspaceRequest(BaseModel):
     """Typed MCP request payload for saving a draft workspace as an artifact."""
 

src/wf_mcp/workflow_surface/tools.py

@@ -12,6 +12,8 @@ from .handlers import WorkflowSurfaceHandlers
 from .models import (
     CallCapabilityResult,
     CreateArtifactFromWorkspaceRequest,
+    CreateDraftWorkspaceFromCapabilityRequest,
+    CreateDraftWorkspaceFromCapabilityResult,
     CreateDraftWorkspaceRequest,
     CreateMinimalDraftWorkspaceRequest,
     CreateWrapperFromWorkspaceRequest,
@@ -395,6 +397,32 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
             )
         )
 
+    @server.tool(
+        name="wf.workflow.create_draft_workspace_from_capability",
+        title="Create Draft Workspace From Capability",
+        description=(
+            "Inspect one capability, use its wrapper_hints to bootstrap a "
+            "patchable draft workspace, and return the hints used."
+        ),
+    )
+    async def create_draft_workspace_from_capability(
+        request: CreateDraftWorkspaceFromCapabilityRequest,
+    ) -> CreateDraftWorkspaceFromCapabilityResult:
+        return CreateDraftWorkspaceFromCapabilityResult.model_validate(
+            await handlers.create_draft_workspace_from_capability(
+                workspace_id=request.workspace_id,
+                capability_name=request.capability_name,
+                name=request.name,
+                title=request.title,
+                input_schema=request.input_schema,
+                state_schema=request.state_schema,
+                output_schema=request.output_schema,
+                input_map=request.input_map,
+                output_map=request.output_map,
+                error_message_source=request.error_message_source,
+            )
+        )
+
     @server.tool(
         name="wf.workflow.create_artifact_from_workspace",
         title="Create Workflow Artifact From Workspace",

tests/wf_mcp/test_server.py

@@ -115,6 +115,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.set_step_input_map" in names
             assert "wf.workflow.set_step_output_map" in names
             assert "wf.workflow.create_minimal_draft_workspace" in names
+            assert "wf.workflow.create_draft_workspace_from_capability" in names
             assert "wf.workflow.create_artifact_from_workspace" in names
             assert "wf.workflow.create_wrapper_from_workspace" in names
             assert "wf.workflow.run_deployment" in names
@@ -142,6 +143,18 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
                 == "Public input JSON Schema for the workflow or wrapper being "
                 "drafted."
             )
+            from_capability_input = tools_by_name[
+                "wf.workflow.create_draft_workspace_from_capability"
+            ].inputSchema
+            from_capability_request = from_capability_input["properties"]["request"]
+            assert "capability_name" in from_capability_request["properties"]
+            assert "input_schema" in from_capability_request["properties"]
+            assert "output_map" in from_capability_request["properties"]
+            from_capability_output = tools_by_name[
+                "wf.workflow.create_draft_workspace_from_capability"
+            ].outputSchema
+            assert from_capability_output is not None
+            assert "wrapper_hints" in from_capability_output["properties"]
             wrapper_workspace_input = tools_by_name[
                 "wf.workflow.create_wrapper_from_workspace"
             ].inputSchema

tests/wf_mcp/test_workflow_surface.py

@@ -680,6 +680,38 @@ def test_workflow_surface_creates_minimal_draft_workspace_with_error_route() ->
     assert workspace.draft["steps"]["tool_error"]["in"] == {"state.echoed": "message"}
 
 
+def test_workflow_surface_creates_draft_workspace_from_capability_hints() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_workspace_from_capability"
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_workspace_from_capability_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)
+
+    result = asyncio.run(
+        handlers.create_draft_workspace_from_capability(
+            workspace_id="echo_from_capability",
+            capability_name="demo.personal.echo_tool",
+            name="echo_from_capability",
+        )
+    )
+    assert service.draft_workspace_store is not None
+    workspace = service.draft_workspace_store.get_workspace("echo_from_capability")
+
+    assert result["workspace_id"] == "echo_from_capability"
+    assert result["wrapper_hints"]["input_map"] == {"input.text": "text"}
+    assert result["wrapper_hints"]["output_map"] == {"echoed": "state.echoed"}
+    assert workspace.draft["steps"]["call"]["use"] == "demo.personal.echo_tool"
+    assert workspace.draft["steps"]["call"]["in"] == {"input.text": "text"}
+    assert workspace.draft["steps"]["call"]["out"] == {"echoed": "state.echoed"}
+
+
 def test_workflow_surface_creates_artifact_from_workspace() -> None:
     artifact_store = FileWorkflowArtifactStore(
         local_temp_root() / "surface_workspace_artifact"