call capability with deployment id

docs/workflow_capabilities.md

@@ -223,6 +223,10 @@ A client authoring workflows, including an LLM client, should be able to:
 5. inspect the normalized output and outcome
 6. reuse that capability inside a graph
 
+Saved wrapper artifacts can be called with a deployment id when they use logical
+source names. The deployment supplies the concrete source bindings for that
+test call, matching the way `run_deployment` resolves a full saved workflow.
+
 That direct-call surface is different from:
 
 - calling the raw upstream MCP tool

src/wf_mcp/workflow_surface/handlers.py

@@ -47,11 +47,16 @@ class WorkflowSurfaceHandlers:
         *,
         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)
+            return await self._call_wrapper_artifact(
+                wrapper_artifact,
+                payload,
+                deployment_id=deployment_id,
+            )
 
         spec = self.service._get_qualified_spec(qualified_name)
         handler = build_async_registry(spec)[spec.name]
@@ -83,6 +88,8 @@ class WorkflowSurfaceHandlers:
         self,
         artifact: WorkflowArtifact,
         payload: dict[str, Any],
+        *,
+        deployment_id: str | None,
     ) -> dict[str, Any]:
         """Execute a saved wrapper artifact through the workflow runner."""
         unsupported = _unsupported_interrupt_diagnostic(artifact)
@@ -93,7 +100,25 @@ class WorkflowSurfaceHandlers:
         # Full saved workflows stay on `run_deployment` until core supports
         # graph-as-node semantics instead of us faking subgraphs at this layer.
         plan = _raw_plan_from_artifact(artifact)
-        run = await self.service.run_workflow_from_plan(plan, payload)
+        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),
             "outcome": run.status.value,

src/wf_mcp/workflow_surface/tools.py

@@ -29,16 +29,19 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         title="Call Workflow Capability",
         description=(
             "Execute one planner-visible workflow capability once and return its "
-            "normalized outcome and output."
+            "normalized outcome and output. Pass deployment_id for saved wrappers "
+            "that use deployment-bound logical sources."
         ),
     )
     async def call_capability(
         qualified_name: str,
         payload: dict[str, Any],
+        deployment_id: str | None = None,
     ) -> dict[str, Any]:
         return await handlers.call_capability(
             qualified_name=qualified_name,
             payload=payload,
+            deployment_id=deployment_id,
         )
 
     @server.tool(

tests/wf_mcp/test_workflow_surface.py

@@ -287,6 +287,47 @@ def test_workflow_surface_calls_saved_wrapper_artifact() -> None:
     assert payload["output"]["echoed"] == "hello"
 
 
+def test_workflow_surface_calls_saved_wrapper_artifact_with_deployment_bindings() -> (
+    None
+):
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_wrapper_bound_call"
+    )
+    wrapper = _logical_echo_artifact().model_copy(
+        update={"id": "logical_echo_wrapper", "kind": "wrapper"}
+    )
+    artifact_store.save_artifact(wrapper)
+    artifact_store.save_deployment(
+        WorkflowDeployment(
+            id="logical_echo_wrapper.personal",
+            artifact_id="logical_echo_wrapper",
+            artifact_version=1,
+            bindings={"demo": "demo.personal"},
+        )
+    )
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_wrapper_bound_call_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)
+
+    payload = asyncio.run(
+        handlers.call_capability(
+            qualified_name="workflow.logical_echo_wrapper.v1",
+            payload={"text": "hello"},
+            deployment_id="logical_echo_wrapper.personal",
+        )
+    )
+
+    assert payload["qualified_name"] == "workflow.logical_echo_wrapper.v1"
+    assert payload["outcome"] == "completed"
+    assert payload["output"]["echoed"] == "hello"
+
+
 def _handlers(artifact_store: FileWorkflowArtifactStore) -> WorkflowSurfaceHandlers:
     service = WfMcpService(
         store=FileStore(local_temp_root() / "surface_mcp"),