fifth slice: Capability api!

src/wf_api/__init__.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 from .artifacts import WorkflowArtifactApi
 from .backend import TraceRange, WorkflowApiBackend
+from .capabilities import WorkflowCapabilityApi
 from .constants import (
     DEFAULT_CALL_STEP_ID,
     DEFAULT_ERROR_OUTCOME,
@@ -56,6 +57,7 @@ __all__ = [
     "WorkflowApi",
     "WorkflowApiBackend",
     "WorkflowArtifactApi",
+    "WorkflowCapabilityApi",
     "WorkflowArtifactCataloger",
     "WorkflowDeploymentApi",
     "WorkflowDraftApi",

src/wf_api/capabilities.py

@@ -0,0 +1,444 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from wf_artifacts import (
+    DependencyDiagnostic,
+    DiagnosticSeverity,
+    RequiredCapability,
+    WorkflowArtifact,
+    WorkflowCapabilityRef,
+)
+from wf_authoring import build_async_registry
+from wf_core import RuntimeContext
+from wf_core.models.steps import InputBinding, OutputBinding
+from wf_core.paths import GraphSourcePath
+from wf_platform import CapabilitySource, page_items
+
+from .drafts import WorkflowDraftApi
+from .models import RawWorkflowPlan
+from .next_actions import NextActions
+from .operation_context import WorkflowOperationContext
+from .refs import parse_workflow_surface_capability_id
+from .saved_subgraphs import direct_wrapper_interrupt_diagnostic
+from .wrapper_hints import (
+    workflow_output_schema_for_authoring,
+    wrapper_hints_for_capability,
+)
+
+
+def _matches_query(*values: object, query: str | None) -> bool:
+    if query is None:
+        return True
+    needle = query.strip().casefold()
+    if not needle:
+        return True
+    return any(needle in str(value).casefold() for value in values if value is not None)
+
+
+def _paged_list_payload(
+    key: str,
+    items: Sequence[dict[str, Any]],
+    *,
+    cursor: str | None,
+    limit: int,
+) -> dict[str, Any]:
+    page = page_items(items, cursor=cursor, limit=limit)
+    return {key: list(page.items), "next_cursor": page.next_cursor, "total": page.total}
+
+
+def _schema_field_names(schema: dict[str, Any]) -> list[str]:
+    """Return top-level JSON object property names for compact discovery rows."""
+    properties = schema.get("properties")
+    if not isinstance(properties, dict):
+        return []
+    return sorted(str(name) for name in properties)
+
+
+def _artifact_capability_id(artifact: WorkflowArtifact) -> str:
+    """Use the same stable name shape as workflow artifact catalog entries."""
+    return str(
+        WorkflowCapabilityRef(
+            artifact_id=artifact.id,
+            version=artifact.version,
+        )
+    )
+
+
+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 _source_id_for_capability(
+    sources: Mapping[str, CapabilitySource],
+    qualified_name: str,
+) -> str | None:
+    """Return the source that currently owns one workflow capability."""
+    for source in sources.values():
+        if qualified_name in source.capabilities.node_specs:
+            return source.id
+    return None
+
+
+def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
+    """Validate the stored plan shape expected by the broker workflow runner."""
+    return RawWorkflowPlan.model_validate(
+        {
+            "name": _plan_field(artifact, "name"),
+            "input_schema": _plan_field(artifact, "input_schema"),
+            "state_schema": _plan_field(artifact, "state_schema"),
+            "output_schema": _plan_field(artifact, "output_schema"),
+            "outcomes": artifact.plan.get("outcomes", ["ok"]),
+            "output": artifact.plan.get("output", []),
+            "start": _plan_field(artifact, "start"),
+            "nodes": _plan_field(artifact, "nodes"),
+            "edges": _plan_field(artifact, "edges"),
+        }
+    )
+
+
+def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
+    try:
+        return artifact.plan[field_name]
+    except KeyError as exc:
+        raise ValueError(
+            f"workflow artifact {artifact.id}@{artifact.version} "
+            f"is missing plan field {field_name!r}"
+        ) from exc
+
+
+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("-", "_")
+
+
+class WorkflowCapabilityApi:
+    """Workflow-facing capability discovery, inspection, and REPL calls.
+
+    This service owns the source/wrapper projection, while adapter-specific MCP
+    tool schemas stay outside wf_api.
+    """
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+        self.drafts = WorkflowDraftApi(context)
+
+    async def list_capabilities(
+        self,
+        *,
+        query: str | None = None,
+        source_id: str | None = None,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        """Return compact paged planner-visible workflow capability summaries."""
+        capabilities = [
+            {
+                "name": detail.name,
+                "source_id": source.id,
+                "kind": "node_spec",
+                "description": detail.description,
+                "outcomes": list(detail.outcomes),
+                "is_async": detail.is_async,
+                "input_fields": _schema_field_names(detail.input_schema),
+                "output_fields": _schema_field_names(
+                    workflow_output_schema_for_authoring(detail.output_schema)
+                ),
+            }
+            for source in sorted(
+                self.context.capability_sources.values(),
+                key=lambda source: source.id,
+            )
+            if source.enabled and source.visibility.planner
+            if source_id is None or source.id == source_id
+            for detail in source.as_inventory().capabilities.node_spec_details
+            if _matches_query(
+                detail.name,
+                detail.description,
+                query=query,
+            )
+        ]
+        capabilities.extend(
+            self._wrapper_capability_summaries(query=query, source_id=source_id)
+        )
+        capabilities.sort(key=lambda capability: capability["name"])
+        return _paged_list_payload(
+            "capabilities",
+            capabilities,
+            cursor=cursor,
+            limit=limit,
+        )
+
+    async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
+        """Return one planner-visible workflow capability contract."""
+        for source in self.context.capability_sources.values():
+            if not source.enabled or not source.visibility.planner:
+                continue
+            for detail in source.as_inventory().capabilities.node_spec_details:
+                if detail.name == qualified_name:
+                    detail_payload = detail.model_dump(mode="json")
+                    detail_payload["wrapper_hints"] = wrapper_hints_for_capability(
+                        capability_name=detail.name,
+                        input_schema=detail.input_schema,
+                        output_schema=detail.output_schema,
+                        outcomes=detail.outcomes,
+                    ).model_dump(mode="json")
+                    return detail_payload
+        wrapper_detail = self._wrapper_capability_detail(qualified_name)
+        if wrapper_detail is not None:
+            return wrapper_detail
+        raise KeyError(f"unknown workflow capability {qualified_name!r}")
+
+    async def call_capability(
+        self,
+        *,
+        qualified_name: str,
+        payload: dict[str, Any],
+        deployment_id: str | None = None,
+    ) -> dict[str, Any]:
+        """Execute one planner-visible workflow capability for authoring tests."""
+        wrapper_artifact = self._wrapper_artifact_for_capability_name(qualified_name)
+        if wrapper_artifact is not None:
+            return await self._call_wrapper_artifact(
+                wrapper_artifact,
+                payload,
+                deployment_id=deployment_id,
+            )
+
+        spec = self.context.specs.get_qualified_spec(qualified_name)
+        handler = build_async_registry(spec)[spec.name]
+        source_id = _source_id_for_capability(
+            self.context.capability_sources,
+            spec.name,
+        )
+        try:
+            result = await handler(payload, RuntimeContext(current_node_id=spec.name))
+        except Exception as exc:
+            return {
+                "qualified_name": spec.name,
+                "source_id": source_id,
+                "kind": "node_spec",
+                "deployment_id": None,
+                "outcome": "runtime_error",
+                "output": None,
+                "diagnostics": [
+                    DependencyDiagnostic(
+                        severity=DiagnosticSeverity.ERROR,
+                        code="capability_call_failed",
+                        logical_ref=spec.name,
+                        bound_source=source_id,
+                        message=(
+                            f"Capability {spec.name!r} failed during test call: {exc}"
+                        ),
+                        repair_hint=(
+                            "Check the source runtime, then retry the capability "
+                            "or inspect the deployment run if this happened inside "
+                            "a workflow."
+                        ),
+                    ).model_dump(mode="json")
+                ],
+            }
+        return {
+            "qualified_name": spec.name,
+            "source_id": source_id,
+            "kind": "node_spec",
+            "deployment_id": None,
+            "outcome": result["outcome"],
+            "output": result["output"],
+            "diagnostics": [],
+        }
+
+    def _wrapper_artifact_for_capability_name(
+        self,
+        qualified_name: str,
+    ) -> WorkflowArtifact | None:
+        """Resolve a saved node-like wrapper artifact from its stable capability name."""
+        try:
+            capability_id = parse_workflow_surface_capability_id(qualified_name)
+        except ValueError:
+            return None
+        if (
+            not isinstance(capability_id, WorkflowCapabilityRef)
+            or self.context.artifact_store is None
+        ):
+            return None
+        try:
+            artifact = self.context.artifact_store.get_artifact(
+                capability_id.artifact_id,
+                capability_id.version,
+            )
+        except KeyError:
+            return None
+        if artifact.kind != "wrapper":
+            return None
+        return artifact
+
+    def _wrapper_capability_summaries(
+        self,
+        *,
+        query: str | None,
+        source_id: str | None,
+    ) -> list[dict[str, Any]]:
+        """Project saved wrappers into workflow capability discovery rows.
+
+        Wrapper artifacts are not live source NodeSpecs, but authors need to
+        discover and test them through the same workflow-facing REPL surface.
+        Full saved workflows stay out of this projection until graph-as-node is
+        real in core.
+        """
+        if source_id not in {None, "workflow"} or self.context.artifact_store is None:
+            return []
+        rows: list[dict[str, Any]] = []
+        for artifact in self.context.artifact_store.list_artifacts():
+            if artifact.kind != "wrapper":
+                continue
+            name = _artifact_capability_id(artifact)
+            if not _matches_query(
+                name,
+                artifact.description,
+                query=query,
+            ):
+                continue
+            rows.append(
+                {
+                    "name": name,
+                    "source_id": "workflow",
+                    "kind": "wrapper_artifact",
+                    "artifact_id": artifact.id,
+                    "version": artifact.version,
+                    "title": artifact.title,
+                    "description": artifact.description,
+                    "outcomes": list(artifact.outcomes),
+                    "is_async": True,
+                    "input_fields": _schema_field_names(artifact.input_schema),
+                    "output_fields": _schema_field_names(artifact.output_schema),
+                }
+            )
+        return rows
+
+    def _wrapper_capability_detail(
+        self,
+        qualified_name: str,
+    ) -> dict[str, Any] | None:
+        """Return a NodeSpec-like contract for one saved wrapper artifact."""
+        artifact = self._wrapper_artifact_for_capability_name(qualified_name)
+        if artifact is None:
+            return None
+        return {
+            "name": _artifact_capability_id(artifact),
+            "source_id": "workflow",
+            "kind": "wrapper_artifact",
+            "artifact_id": artifact.id,
+            "version": artifact.version,
+            "title": artifact.title,
+            "description": artifact.description,
+            "outcomes": list(artifact.outcomes),
+            "is_async": True,
+            "input_schema": artifact.input_schema,
+            "output_schema": artifact.output_schema,
+            "required_capabilities": _required_capability_payloads(
+                artifact.required_capability_map()
+            ),
+            "wrapper_hints": wrapper_hints_for_capability(
+                capability_name=_artifact_capability_id(artifact),
+                input_schema=artifact.input_schema,
+                output_schema=artifact.output_schema,
+                outcomes=list(artifact.outcomes),
+            ).model_dump(mode="json"),
+        }
+
+    async def _call_wrapper_artifact(
+        self,
+        artifact: WorkflowArtifact,
+        payload: dict[str, Any],
+        *,
+        deployment_id: str | None,
+    ) -> dict[str, Any]:
+        """Execute a saved wrapper artifact through the workflow runner."""
+        unsupported = direct_wrapper_interrupt_diagnostic(artifact)
+        if unsupported is not None:
+            raise ValueError(unsupported.message)
+
+        # Direct capability calls remain wrapper-only. Full saved workflows run
+        # through deployments, where native subgraph dependencies and bindings
+        # are prepared before core execution.
+        plan = _raw_plan_from_artifact(artifact)
+        deployment = None
+        if deployment_id is not None:
+            if self.context.artifact_store is None:
+                raise KeyError("workflow artifact store is not configured")
+            deployment = self.context.artifact_store.get_deployment(deployment_id)
+            if (
+                deployment.artifact_id != artifact.id
+                or deployment.artifact_version != artifact.version
+            ):
+                raise ValueError(
+                    f"deployment {deployment_id!r} does not target "
+                    f"workflow.{artifact.id}.v{artifact.version}"
+                )
+        run = await self.context.runtime.run_workflow_from_plan(
+            plan,
+            payload,
+            deployment=deployment,
+            artifact=artifact,
+        )
+        return {
+            "qualified_name": _artifact_capability_id(artifact),
+            "source_id": "workflow",
+            "kind": "wrapper_artifact",
+            "deployment_id": deployment_id,
+            "outcome": run.status.value,
+            "output": run.output,
+            "diagnostics": [],
+        }
+
+    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: Sequence[InputBinding] | None = None,
+        output: Sequence[OutputBinding] | None = None,
+        input_map: dict[str, str] | None = None,
+        output_map: dict[str, str] | None = None,
+        error_message_source: str | GraphSourcePath | 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.drafts.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=input,
+            output=output,
+            input_map=None if input is not None else (input_map or hints["input_map"]),
+            output_map=None
+            if output is not None
+            else (output_map or hints["output_map"]),
+            error_message_source=error_message_source,
+            title=title,
+        )
+        return {
+            **result,
+            "wrapper_hints": hints,
+            "next_actions": NextActions.from_wrapper_hints(
+                workspace_id=workspace_id,
+                revision=int(result["revision"]),
+                hints=hints,
+            ).model_dump(mode="json"),
+        }

src/wf_api/operation_context.py

@@ -13,6 +13,7 @@ from wf_artifacts import (
     WorkflowArtifactStore,
     WorkflowDeployment,
 )
+from wf_authoring import NodeSpec
 from wf_core import RunState
 from wf_platform import CapabilitySource
 
@@ -46,7 +47,7 @@ class WorkflowSpecProvider(Protocol):
         """Planner-visible capability sources keyed by source id."""
         ...
 
-    def get_qualified_spec(self, qualified_name: str) -> object:
+    def get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:
         """Return the node spec for one fully qualified capability name."""
         ...
 

src/wf_mcp/workflow_surface/handlers.py

@@ -5,18 +5,8 @@ from typing import TYPE_CHECKING, Any
 
 from wf_artifacts import (
     ArtifactKind,
-    DependencyDiagnostic,
-    DiagnosticSeverity,
     DraftWorkspaceStore,
-    RequiredCapability,
-    WorkflowArtifact,
-    WorkflowCapabilityRef,
 )
-from wf_platform import (
-    CapabilitySource,
-)
-from wf_authoring import build_async_registry
-from wf_core import RuntimeContext
 from wf_core.models.steps import (
     InputBinding,
     OutputBinding,
@@ -24,22 +14,14 @@ from wf_core.models.steps import (
 from wf_core.paths import GraphSourcePath
 
 from wf_api.artifacts import WorkflowArtifactApi
+from wf_api.capabilities import WorkflowCapabilityApi
 from wf_api.deployments import WorkflowDeploymentApi
 from wf_api.drafts import WorkflowDraftApi
 from wf_api.models import RawWorkflowPlan
-from wf_api.next_actions import NextActions
-from wf_api.refs import parse_workflow_surface_capability_id
 from wf_api.runs import WorkflowRunApi
-from wf_api.saved_subgraphs import (
-    direct_wrapper_interrupt_diagnostic,
-)
-from wf_api.wrapper_hints import (
-    workflow_output_schema_for_authoring,
-    wrapper_hints_for_capability,
-)
 
 from ..broker.service.workflow_operation_context import context_from_service
-from ..shared import matches_query, paged_list_payload
+from ..shared import paged_list_payload
 from .models import TraceRange
 
 if TYPE_CHECKING:
@@ -52,6 +34,7 @@ class WorkflowSurfaceHandlers:
     def __init__(self, service: WfMcpService) -> None:
         self.service = service
         context = context_from_service(service)
+        self._capabilities = WorkflowCapabilityApi(context)
         self._drafts = WorkflowDraftApi(context)
         self._artifacts = WorkflowArtifactApi(context)
         self._deployments = WorkflowDeploymentApi(context)
@@ -88,62 +71,18 @@ class WorkflowSurfaceHandlers:
         limit: int = 50,
     ) -> dict[str, Any]:
         """Return compact paged planner-visible workflow capability summaries."""
-        capabilities = [
+        return await self._capabilities.list_capabilities(
-            {
-                "name": detail.name,
-                "source_id": source.id,
-                "kind": "node_spec",
-                "description": detail.description,
-                "outcomes": list(detail.outcomes),
-                "is_async": detail.is_async,
-                "input_fields": _schema_field_names(detail.input_schema),
-                "output_fields": _schema_field_names(
-                    workflow_output_schema_for_authoring(detail.output_schema)
-                ),
-            }
-            for source in sorted(
-                self.service.capability_sources.values(),
-                key=lambda source: source.id,
-            )
-            if source.enabled and source.visibility.planner
-            if source_id is None or source.id == source_id
-            for detail in source.as_inventory().capabilities.node_spec_details
-            if matches_query(
-                detail.name,
-                detail.description,
             query=query,
-            )
+            source_id=source_id,
-        ]
-        capabilities.extend(
-            self._wrapper_capability_summaries(query=query, source_id=source_id)
-        )
-        capabilities.sort(key=lambda capability: capability["name"])
-        return paged_list_payload(
-            "capabilities",
-            capabilities,
             cursor=cursor,
             limit=limit,
         )
 
     async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
         """Return one planner-visible workflow capability contract."""
-        for source in self.service.capability_sources.values():
+        return await self._capabilities.inspect_capability(
-            if not source.enabled or not source.visibility.planner:
+            qualified_name=qualified_name,
-                continue
+        )
-            for detail in source.as_inventory().capabilities.node_spec_details:
-                if detail.name == qualified_name:
-                    detail_payload = detail.model_dump(mode="json")
-                    detail_payload["wrapper_hints"] = wrapper_hints_for_capability(
-                        capability_name=detail.name,
-                        input_schema=detail.input_schema,
-                        output_schema=detail.output_schema,
-                        outcomes=detail.outcomes,
-                    ).model_dump(mode="json")
-                    return detail_payload
-        wrapper_detail = self._wrapper_capability_detail(qualified_name)
-        if wrapper_detail is not None:
-            return wrapper_detail
-        raise KeyError(f"unknown workflow capability {qualified_name!r}")
 
     async def call_capability(
         self,
@@ -153,201 +92,12 @@ class WorkflowSurfaceHandlers:
         deployment_id: str | None = None,
     ) -> dict[str, Any]:
         """Execute one planner-visible workflow capability for authoring tests."""
-        wrapper_artifact = self._wrapper_artifact_for_capability_name(qualified_name)
+        return await self._capabilities.call_capability(
-        if wrapper_artifact is not None:
+            qualified_name=qualified_name,
-            return await self._call_wrapper_artifact(
+            payload=payload,
-                wrapper_artifact,
-                payload,
             deployment_id=deployment_id,
         )
 
-        spec = self.service._get_qualified_spec(qualified_name)
-        handler = build_async_registry(spec)[spec.name]
-        source_id = _source_id_for_capability(
-            self.service.capability_sources,
-            spec.name,
-        )
-        try:
-            result = await handler(payload, RuntimeContext(current_node_id=spec.name))
-        except Exception as exc:
-            return {
-                "qualified_name": spec.name,
-                "source_id": source_id,
-                "kind": "node_spec",
-                "deployment_id": None,
-                "outcome": "runtime_error",
-                "output": None,
-                "diagnostics": [
-                    DependencyDiagnostic(
-                        severity=DiagnosticSeverity.ERROR,
-                        code="capability_call_failed",
-                        logical_ref=spec.name,
-                        bound_source=source_id,
-                        message=(
-                            f"Capability {spec.name!r} failed during test call: {exc}"
-                        ),
-                        repair_hint=(
-                            "Check the source runtime, then retry the capability "
-                            "or inspect the deployment run if this happened inside "
-                            "a workflow."
-                        ),
-                    ).model_dump(mode="json")
-                ],
-            }
-        return {
-            "qualified_name": spec.name,
-            "source_id": source_id,
-            "kind": "node_spec",
-            "deployment_id": None,
-            "outcome": result["outcome"],
-            "output": result["output"],
-            "diagnostics": [],
-        }
-
-    def _wrapper_artifact_for_capability_name(
-        self,
-        qualified_name: str,
-    ) -> WorkflowArtifact | None:
-        """Resolve a saved node-like wrapper artifact from its stable capability name."""
-        try:
-            capability_id = parse_workflow_surface_capability_id(qualified_name)
-        except ValueError:
-            return None
-        if (
-            not isinstance(capability_id, WorkflowCapabilityRef)
-            or self.service.artifact_store is None
-        ):
-            return None
-        try:
-            artifact = self.service.artifact_store.get_artifact(
-                capability_id.artifact_id,
-                capability_id.version,
-            )
-        except KeyError:
-            return None
-        if artifact.kind != "wrapper":
-            return None
-        return artifact
-
-    def _wrapper_capability_summaries(
-        self,
-        *,
-        query: str | None,
-        source_id: str | None,
-    ) -> list[dict[str, Any]]:
-        """Project saved wrappers into workflow capability discovery rows.
-
-        Wrapper artifacts are not live source NodeSpecs, but authors need to
-        discover and test them through the same workflow-facing REPL surface.
-        Full saved workflows stay out of this projection until graph-as-node is
-        real in core.
-        """
-        if source_id not in {None, "workflow"} or self.service.artifact_store is None:
-            return []
-        rows: list[dict[str, Any]] = []
-        for artifact in self.service.artifact_store.list_artifacts():
-            if artifact.kind != "wrapper":
-                continue
-            name = _artifact_capability_id(artifact)
-            if not matches_query(
-                name,
-                artifact.description,
-                query=query,
-            ):
-                continue
-            rows.append(
-                {
-                    "name": name,
-                    "source_id": "workflow",
-                    "kind": "wrapper_artifact",
-                    "artifact_id": artifact.id,
-                    "version": artifact.version,
-                    "title": artifact.title,
-                    "description": artifact.description,
-                    "outcomes": list(artifact.outcomes),
-                    "is_async": True,
-                    "input_fields": _schema_field_names(artifact.input_schema),
-                    "output_fields": _schema_field_names(artifact.output_schema),
-                }
-            )
-        return rows
-
-    def _wrapper_capability_detail(
-        self,
-        qualified_name: str,
-    ) -> dict[str, Any] | None:
-        """Return a NodeSpec-like contract for one saved wrapper artifact."""
-        artifact = self._wrapper_artifact_for_capability_name(qualified_name)
-        if artifact is None:
-            return None
-        return {
-            "name": _artifact_capability_id(artifact),
-            "source_id": "workflow",
-            "kind": "wrapper_artifact",
-            "artifact_id": artifact.id,
-            "version": artifact.version,
-            "title": artifact.title,
-            "description": artifact.description,
-            "outcomes": list(artifact.outcomes),
-            "is_async": True,
-            "input_schema": artifact.input_schema,
-            "output_schema": artifact.output_schema,
-            "required_capabilities": _required_capability_payloads(
-                artifact.required_capability_map()
-            ),
-            "wrapper_hints": wrapper_hints_for_capability(
-                capability_name=_artifact_capability_id(artifact),
-                input_schema=artifact.input_schema,
-                output_schema=artifact.output_schema,
-                outcomes=list(artifact.outcomes),
-            ).model_dump(mode="json"),
-        }
-
-    async def _call_wrapper_artifact(
-        self,
-        artifact: WorkflowArtifact,
-        payload: dict[str, Any],
-        *,
-        deployment_id: str | None,
-    ) -> dict[str, Any]:
-        """Execute a saved wrapper artifact through the workflow runner."""
-        unsupported = direct_wrapper_interrupt_diagnostic(artifact)
-        if unsupported is not None:
-            raise ValueError(unsupported.message)
-
-        # Direct capability calls remain wrapper-only. Full saved workflows run
-        # through deployments, where native subgraph dependencies and bindings
-        # are prepared before core execution.
-        plan = _raw_plan_from_artifact(artifact)
-        deployment = None
-        if deployment_id is not None:
-            if self.service.artifact_store is None:
-                raise KeyError("workflow artifact store is not configured")
-            deployment = self.service.artifact_store.get_deployment(deployment_id)
-            if (
-                deployment.artifact_id != artifact.id
-                or deployment.artifact_version != artifact.version
-            ):
-                raise ValueError(
-                    f"deployment {deployment_id!r} does not target "
-                    f"workflow.{artifact.id}.v{artifact.version}"
-                )
-        run = await self.service.run_workflow_from_plan(
-            plan,
-            payload,
-            deployment=deployment,
-            artifact=artifact,
-        )
-        return {
-            "qualified_name": _artifact_capability_id(artifact),
-            "source_id": "workflow",
-            "kind": "wrapper_artifact",
-            "deployment_id": deployment_id,
-            "outcome": run.status.value,
-            "output": run.output,
-            "diagnostics": [],
-        }
-
     async def save_artifact(self, artifact: dict[str, Any]) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
@@ -587,33 +337,20 @@ class WorkflowSurfaceHandlers:
         error_message_source: str | GraphSourcePath | None = None,
     ) -> dict[str, Any]:
         """Create a patchable draft workspace from inspect_capability hints."""
-        capability = await self.inspect_capability(qualified_name=capability_name)
+        return await self._capabilities.create_draft_workspace_from_capability(
-        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"],
+            name=name,
-            state_schema=state_schema or hints["state_schema"],
+            title=title,
-            output_schema=output_schema or hints["output_schema"],
+            input_schema=input_schema,
+            state_schema=state_schema,
+            output_schema=output_schema,
             input=input,
             output=output,
-            input_map=None if input is not None else (input_map or hints["input_map"]),
+            input_map=input_map,
-            output_map=None
+            output_map=output_map,
-            if output is not None
-            else (output_map or hints["output_map"]),
             error_message_source=error_message_source,
-            title=title,
         )
-        return {
-            **result,
-            "wrapper_hints": hints,
-            "next_actions": NextActions.from_wrapper_hints(
-                workspace_id=workspace_id,
-                revision=int(result["revision"]),
-                hints=hints,
-            ).model_dump(mode="json"),
-        }
 
     async def create_artifact_from_workspace(
         self,
@@ -764,84 +501,3 @@ class WorkflowSurfaceHandlers:
             run_id=run_id,
             trace_range=trace_range,
         )
-
-
-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 _schema_field_names(schema: dict[str, Any]) -> list[str]:
-    """Return top-level JSON object property names for compact discovery rows."""
-    properties = schema.get("properties")
-    if not isinstance(properties, dict):
-        return []
-    return sorted(str(name) for name in properties)
-
-
-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,
-) -> str | None:
-    """Return the source that currently owns one workflow capability."""
-    for source in sources.values():
-        if qualified_name in source.capabilities.node_specs:
-            return source.id
-    return None
-
-
-def _capability_name(qualified_name: str) -> str | None:
-    """Return the local name of one qualified capability ref if it is valid."""
-    try:
-        parsed = parse_workflow_surface_capability_id(qualified_name)
-    except ValueError:
-        return None
-    if isinstance(parsed, WorkflowCapabilityRef):
-        return None
-    return parsed.name
-
-
-def _artifact_capability_id(artifact: WorkflowArtifact) -> str:
-    """Use the same stable name shape as workflow artifact catalog entries."""
-    return str(
-        WorkflowCapabilityRef(
-            artifact_id=artifact.id,
-            version=artifact.version,
-        )
-    )
-
-
-def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
-    """Validate the stored plan shape expected by the broker workflow runner."""
-    return RawWorkflowPlan.model_validate(
-        {
-            "name": _plan_field(artifact, "name"),
-            "input_schema": _plan_field(artifact, "input_schema"),
-            "state_schema": _plan_field(artifact, "state_schema"),
-            "output_schema": _plan_field(artifact, "output_schema"),
-            "outcomes": artifact.plan.get("outcomes", ["ok"]),
-            "output": artifact.plan.get("output", []),
-            "start": _plan_field(artifact, "start"),
-            "nodes": _plan_field(artifact, "nodes"),
-            "edges": _plan_field(artifact, "edges"),
-        }
-    )
-
-
-def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
-    try:
-        return artifact.plan[field_name]
-    except KeyError as exc:
-        raise ValueError(
-            f"workflow artifact {artifact.id}@{artifact.version} "
-            f"is missing plan field {field_name!r}"
-        ) from exc

tests/wf_api/test_capability_api.py

@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import asyncio
+
+from wf_artifacts import FileWorkflowArtifactStore
+from wf_api.capabilities import WorkflowCapabilityApi
+from wf_mcp.broker import WfMcpService
+from wf_mcp.models import ConnectionConfig
+from wf_mcp.storage import FileStore
+from wf_mcp.workflow_surface import WorkflowSurfaceHandlers
+from wf_mcp.broker.service.workflow_operation_context import context_from_service
+
+from tests.wf_mcp.test_support import echo_tool, local_temp_root
+from tests.wf_mcp.workflow_surface.conftest import echo_artifact, failing_tool
+
+
+def _capability_api(
+    artifact_store: FileWorkflowArtifactStore,
+    *,
+    register_echo: bool = False,
+    register_failing: bool = False,
+) -> tuple[WorkflowCapabilityApi, WfMcpService]:
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "caps_mcp" / str(id(artifact_store))),
+        artifact_store=artifact_store,
+    )
+    if register_echo:
+        service.register_connection(
+            ConnectionConfig(id="demo.personal", server="demo", account="personal")
+        )
+        service.register_specs("demo.personal", echo_tool)
+    if register_failing:
+        service.register_connection(
+            ConnectionConfig(id="demo.personal", server="demo", account="personal")
+        )
+        service.register_specs("demo.personal", failing_tool)
+    context = context_from_service(service)
+    return WorkflowCapabilityApi(context), service
+
+
+def test_list_capabilities_returns_planner_visible_sources() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "cap_api_list")
+    api, _service = _capability_api(artifact_store, register_echo=True)
+
+    result = asyncio.run(api.list_capabilities())
+
+    assert result["total"] >= 1
+    assert any(
+        item["name"] == "demo.personal.echo_tool" for item in result["capabilities"]
+    )
+    first = next(
+        item
+        for item in result["capabilities"]
+        if item["name"] == "demo.personal.echo_tool"
+    )
+    assert first["kind"] == "node_spec"
+    assert "input_fields" in first
+    assert "output_fields" in first
+
+
+def test_list_capabilities_filters_by_source() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "cap_api_list_filter"
+    )
+    api, _service = _capability_api(artifact_store, register_echo=True)
+
+    result = asyncio.run(api.list_capabilities(source_id="wf.std", query="truthy"))
+
+    assert [item["name"] for item in result["capabilities"]] == ["wf.std.truthy"]
+
+
+def test_inspect_capability_returns_detail_with_wrapper_hints() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "cap_api_inspect")
+    api, _service = _capability_api(artifact_store, register_echo=True)
+
+    detail = asyncio.run(
+        api.inspect_capability(qualified_name="demo.personal.echo_tool")
+    )
+
+    assert detail["name"] == "demo.personal.echo_tool"
+    assert "wrapper_hints" in detail
+    hints = detail["wrapper_hints"]
+    assert hints["capability_name"] == "demo.personal.echo_tool"
+    assert hints["input_map"] == {"input.text": "text"}
+    assert hints["output_map"] == {"echoed": "state.echoed"}
+
+
+def test_inspect_capability_raises_on_unknown() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "cap_api_inspect_unknown"
+    )
+    api, _service = _capability_api(artifact_store, register_echo=True)
+
+    try:
+        asyncio.run(api.inspect_capability(qualified_name="no.such.capability"))
+        assert False, "expected KeyError"
+    except KeyError as exc:
+        assert "no.such.capability" in str(exc)
+
+
+def test_call_capability_node_spec_success() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "cap_api_call")
+    api, _service = _capability_api(artifact_store, register_echo=True)
+
+    result = asyncio.run(
+        api.call_capability(
+            qualified_name="demo.personal.echo_tool",
+            payload={"text": "hello"},
+        )
+    )
+
+    assert result["kind"] == "node_spec"
+    assert result["outcome"] == "ok"
+    assert result["output"] == {"echoed": "hello"}
+    assert result["diagnostics"] == []
+    assert result["deployment_id"] is None
+
+
+def test_call_capability_node_spec_failure() -> None:
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "cap_api_call_fail")
+    api, _service = _capability_api(artifact_store, register_failing=True)
+
+    result = asyncio.run(
+        api.call_capability(
+            qualified_name="demo.personal.failing_tool",
+            payload={"message": "boom"},
+        )
+    )
+
+    assert result["kind"] == "node_spec"
+    assert result["outcome"] == "runtime_error"
+    assert result["output"] is None
+    assert result["diagnostics"][0]["code"] == "capability_call_failed"
+
+
+def test_list_capabilities_includes_saved_wrapper() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "cap_api_wrapper_list"
+    )
+    artifact_store.save_artifact(
+        echo_artifact().model_copy(
+            update={
+                "id": "echo_wrapper",
+                "kind": "wrapper",
+                "description": "Reusable echo wrapper.",
+            }
+        )
+    )
+    api, _service = _capability_api(artifact_store)
+
+    result = asyncio.run(api.list_capabilities(source_id="workflow", query="echo"))
+
+    names = [item["name"] for item in result["capabilities"]]
+    assert names == ["workflow.echo_wrapper.v1"]
+    row = result["capabilities"][0]
+    assert row["source_id"] == "workflow"
+    assert row["kind"] == "wrapper_artifact"
+    assert row["artifact_id"] == "echo_wrapper"
+    assert row["version"] == 1
+
+
+def test_inspect_capability_saved_wrapper() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "cap_api_wrapper_inspect"
+    )
+    artifact_store.save_artifact(
+        echo_artifact().model_copy(update={"id": "echo_wrapper", "kind": "wrapper"})
+    )
+    api, _service = _capability_api(artifact_store)
+
+    detail = asyncio.run(
+        api.inspect_capability(qualified_name="workflow.echo_wrapper.v1")
+    )
+
+    assert detail["name"] == "workflow.echo_wrapper.v1"
+    assert detail["source_id"] == "workflow"
+    assert detail["kind"] == "wrapper_artifact"
+    assert detail["artifact_id"] == "echo_wrapper"
+    assert detail["outcomes"] == ["completed"]
+    hints = detail["wrapper_hints"]
+    assert hints["capability_name"] == "workflow.echo_wrapper.v1"
+
+
+def test_call_capability_saved_wrapper() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "cap_api_wrapper_call"
+    )
+    artifact_store.save_artifact(
+        echo_artifact().model_copy(update={"id": "echo_wrapper", "kind": "wrapper"})
+    )
+    api, service = _capability_api(artifact_store, register_echo=True)
+
+    result = asyncio.run(
+        api.call_capability(
+            qualified_name="workflow.echo_wrapper.v1",
+            payload={"text": "hi"},
+        )
+    )
+
+    assert result["kind"] == "wrapper_artifact"
+    assert result["outcome"] == "completed"
+    assert result["diagnostics"] == []
+
+
+def test_create_draft_workspace_from_capability() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "cap_api_draft_bootstrap"
+    )
+    api, _service = _capability_api(artifact_store, register_echo=True)
+
+    result = asyncio.run(
+        api.create_draft_workspace_from_capability(
+            workspace_id="echo_ws",
+            capability_name="demo.personal.echo_tool",
+        )
+    )
+
+    assert result["workspace_id"] == "echo_ws"
+    assert result["revision"] == 1
+    assert "wrapper_hints" in result
+    assert "next_actions" in result
+    assert result["wrapper_hints"]["capability_name"] == "demo.personal.echo_tool"
+
+    fetched = asyncio.run(
+        api.drafts.get_draft_workspace(workspace_id="echo_ws", include_draft=True)
+    )
+    assert fetched["draft"]["steps"]["call"]["use"] == "demo.personal.echo_tool"
+
+
+def test_handler_delegates_to_capability_api() -> None:
+    """WorkflowSurfaceHandlers methods produce the same result as direct API."""
+    artifact_store = FileWorkflowArtifactStore(local_temp_root() / "cap_api_delegation")
+    service = WfMcpService(
+        store=FileStore(artifact_store.root / "delegation_mcp"),
+        artifact_store=artifact_store,
+    )
+    service.register_connection(
+        ConnectionConfig(id="demo.personal", server="demo", account="personal")
+    )
+    service.register_specs("demo.personal", echo_tool)
+
+    h = WorkflowSurfaceHandlers(service)
+    context = context_from_service(service)
+    api = WorkflowCapabilityApi(context)
+
+    handler_result = asyncio.run(
+        h.inspect_capability(qualified_name="demo.personal.echo_tool")
+    )
+    api_result = asyncio.run(
+        api.inspect_capability(qualified_name="demo.personal.echo_tool")
+    )
+
+    assert handler_result["name"] == api_result["name"]
+    assert handler_result["wrapper_hints"] == api_result["wrapper_hints"]