call capability result!

docs/wf_mcp_operator_manual.md

@@ -179,6 +179,11 @@ nodes rather than source ownership. Its rows include `source_id`, outcomes, and
 top-level input/output field names, while full JSON schemas stay behind
 `wf.workflow.inspect_capability`.
 
+`wf.workflow.call_capability` is the REPL-style test step. Its result is
+self-describing: `kind` is either `node_spec` or `wrapper_artifact`,
+`source_id` identifies the owner when applicable, and `diagnostics` is empty for
+successful calls.
+
 ### 4. Manage Saved Workflows
 
 Use `wf.workflow.*` for artifacts and deployments:

docs/workflow_capabilities.md

@@ -268,6 +268,8 @@ authoring loop:
   - returns one full workflow capability contract with schemas and outcomes
 - `wf.workflow.call_capability`
   - executes one such capability once for direct testing
+  - returns `qualified_name`, `source_id`, `kind`, optional `deployment_id`,
+    `outcome`, `output`, and `diagnostics`
 
 These are authoring-plane tools. They do not replace the privileged
 `wf.admin.list_sources` source inventory, and older planner-catalog projections
@@ -281,6 +283,8 @@ Recommended discovery order:
    workflow-ready node specs.
 3. Use `wf.workflow.inspect_capability` only for the selected capability's full
    schema contract.
+4. Use `wf.workflow.call_capability` with a plain input object to test the
+   selected contract once before composing it into a draft.
 
 ## Relationship To Capability Sources
 

src/wf_mcp/workflow_surface/handlers.py

@@ -19,7 +19,13 @@ from wf_artifacts import (
     validate_workflow_draft,
     validate_deployment_dependencies,
 )
-from wf_platform import CapabilityRef, NodeSpecInventory, hash_json_schema, page_items
+from wf_platform import (
+    CapabilityRef,
+    CapabilitySource,
+    NodeSpecInventory,
+    hash_json_schema,
+    page_items,
+)
 from wf_authoring import build_async_registry
 from wf_core import RuntimeContext
 
@@ -118,8 +124,15 @@ class WorkflowSurfaceHandlers:
         result = await handler(payload, RuntimeContext(current_node_id=spec.name))
         return {
             "qualified_name": spec.name,
+            "source_id": _source_id_for_capability(
+                self.service.capability_sources,
+                spec.name,
+            ),
+            "kind": "node_spec",
+            "deployment_id": None,
             "outcome": result["outcome"],
             "output": result["output"],
+            "diagnostics": [],
         }
 
     def _wrapper_artifact_for_capability_name(
@@ -176,8 +189,12 @@ class WorkflowSurfaceHandlers:
         )
         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]:
@@ -566,6 +583,17 @@ def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     return sorted(str(name) for name in properties)
 
 
+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:

src/wf_mcp/workflow_surface/models.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+
+class CallCapabilityResult(BaseModel):
+    """Inspector-visible response contract for testing one workflow capability."""
+
+    qualified_name: str = Field(description="Capability name that was executed.")
+    source_id: str | None = Field(
+        default=None,
+        description="Capability source that owned the executed node spec.",
+    )
+    kind: Literal["node_spec", "wrapper_artifact"] = Field(
+        description=(
+            "Indicates whether this call executed a live NodeSpec or a saved "
+            "wrapper artifact."
+        )
+    )
+    deployment_id: str | None = Field(
+        default=None,
+        description="Deployment used to resolve logical bindings, if any.",
+    )
+    outcome: str = Field(description="Workflow outcome returned by the capability.")
+    output: dict[str, Any] | None = Field(
+        default=None,
+        description="Normalized workflow-facing output payload.",
+    )
+    diagnostics: list[dict[str, Any]] = Field(
+        default_factory=list,
+        description="Structured diagnostics. Empty for successful calls.",
+    )

src/wf_mcp/workflow_surface/tools.py

@@ -9,6 +9,7 @@ from wf_artifacts.models import RequiredCapability
 from wf_mcp.broker.service import WfMcpService
 
 from .handlers import WorkflowSurfaceHandlers
+from .models import CallCapabilityResult
 
 
 def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None:
@@ -62,11 +63,13 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         qualified_name: str,
         payload: dict[str, Any],
         deployment_id: str | None = None,
-    ) -> dict[str, Any]:
+    ) -> CallCapabilityResult:
-        return await handlers.call_capability(
+        return CallCapabilityResult.model_validate(
-            qualified_name=qualified_name,
+            await handlers.call_capability(
-            payload=payload,
+                qualified_name=qualified_name,
-            deployment_id=deployment_id,
+                payload=payload,
+                deployment_id=deployment_id,
+            )
         )
 
     @server.tool(

tests/wf_mcp/test_server.py

@@ -42,6 +42,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
         async with client:
             tools = await client.list_tools()
             names = [tool.name for tool in tools]
+            tools_by_name = {tool.name: tool for tool in tools}
             assert "fixture.personal.echo_tool" in names
             assert "wf.admin.list_connections" in names
             assert "wf.admin.get_connection_statuses" in names
@@ -64,6 +65,13 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.workflow.create_artifact_from_draft" in names
             assert "wf.workflow.patch_draft" in names
             assert "wf.workflow.run_deployment" in names
+            call_capability_schema = tools_by_name[
+                "wf.workflow.call_capability"
+            ].outputSchema
+            assert call_capability_schema is not None
+            assert "source_id" in call_capability_schema["properties"]
+            assert "kind" in call_capability_schema["properties"]
+            assert "diagnostics" in call_capability_schema["properties"]
 
             echo_result = await client.call_tool(
                 "fixture.personal.echo_tool",

tests/wf_mcp/test_workflow_surface.py

@@ -614,10 +614,41 @@ def test_workflow_surface_calls_saved_wrapper_artifact() -> None:
     )
 
     assert payload["qualified_name"] == "workflow.echo_wrapper.v1"
+    assert payload["source_id"] == "workflow"
+    assert payload["kind"] == "wrapper_artifact"
+    assert payload["diagnostics"] == []
     assert payload["outcome"] == "completed"
     assert payload["output"]["echoed"] == "hello"
 
 
+def test_workflow_surface_calls_live_node_spec_with_self_describing_response() -> None:
+    service = WfMcpService(
+        store=FileStore(local_temp_root() / "surface_live_capability_call"),
+        artifact_store=FileWorkflowArtifactStore(
+            local_temp_root() / "surface_live_capability_call_artifacts"
+        ),
+    )
+    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="demo.personal.echo_tool",
+            payload={"text": "hello"},
+        )
+    )
+
+    assert payload["qualified_name"] == "demo.personal.echo_tool"
+    assert payload["source_id"] == "demo.personal"
+    assert payload["kind"] == "node_spec"
+    assert payload["diagnostics"] == []
+    assert payload["outcome"] == "ok"
+    assert payload["output"]["echoed"] == "hello"
+
+
 def test_workflow_surface_calls_saved_wrapper_artifact_with_deployment_bindings() -> (
     None
 ):
@@ -655,6 +686,10 @@ def test_workflow_surface_calls_saved_wrapper_artifact_with_deployment_bindings(
     )
 
     assert payload["qualified_name"] == "workflow.logical_echo_wrapper.v1"
+    assert payload["source_id"] == "workflow"
+    assert payload["kind"] == "wrapper_artifact"
+    assert payload["deployment_id"] == "logical_echo_wrapper.personal"
+    assert payload["diagnostics"] == []
     assert payload["outcome"] == "completed"
     assert payload["output"]["echoed"] == "hello"