feat: add cap call rpc command

src/wf_api/surface.py

@@ -26,6 +26,14 @@ class WorkflowCapabilitySurface(Protocol):
         qualified_name: str,
     ) -> dict[str, Any]: ...
 
+    async def call_capability(
+        self,
+        *,
+        qualified_name: str,
+        payload: dict[str, Any],
+        deployment_id: str | None = None,
+    ) -> dict[str, Any]: ...
+
 
 class WorkflowDraftSurface(Protocol):
     """Draft workspace methods exposed by workflow frontends.

src/wf_cli/commands/caps.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
+from pathlib import Path
 from typing import Annotated
 
 import typer
 
 from wf_cli.context import load_cli_context_from_typer
 from wf_cli.formats import ListOutputFormat, emit_list_payload
-from wf_cli.io import emit_json
+from wf_cli.io import CliInputError, emit_json, parse_json_input
 from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
@@ -68,3 +69,47 @@ def inspect_capability(
         context.handlers.inspect_capability(qualified_name=qualified_name),
     )
     emit_json(payload)
+
+
+@app.command("call")
+def call_capability(
+    ctx: typer.Context,
+    qualified_name: Annotated[str, typer.Argument(help="Workflow capability name.")],
+    input_json: Annotated[
+        str | None,
+        typer.Option("--input", help="JSON object payload for the capability."),
+    ] = None,
+    input_file: Annotated[
+        Path | None,
+        typer.Option(
+            "--input-file",
+            exists=True,
+            dir_okay=False,
+            readable=True,
+            help="Read capability JSON object payload from a file.",
+        ),
+    ] = None,
+    deployment_id: Annotated[
+        str | None,
+        typer.Option(
+            "--deployment",
+            help="Deployment id for saved wrappers with deployment-bound sources.",
+        ),
+    ] = None,
+) -> None:
+    """Call one workflow capability once for authoring/runtime smoke tests."""
+    try:
+        payload = parse_json_input(input_json=input_json, input_file=input_file)
+    except CliInputError as exc:
+        raise typer.BadParameter(str(exc)) from exc
+
+    context = load_cli_context_from_typer(ctx)
+    result = run_cli_operation(
+        context,
+        context.handlers.call_capability(
+            qualified_name=qualified_name,
+            payload=payload,
+            deployment_id=deployment_id,
+        ),
+    )
+    emit_json(result)

src/wf_transport_rpc_http/__init__.py

@@ -5,6 +5,7 @@ from .client import RpcWorkflowApiClient
 from .errors import WorkflowRpcError
 from .models import (
     AdminEmptyParams,
+    CallCapabilityParams,
     CreateArtifactFromWorkspaceParams,
     CreateDraftFromCapabilityParams,
     CreateWrapperFromWorkspaceParams,
@@ -37,6 +38,7 @@ from .models import (
 __all__ = [
     "CreateArtifactFromWorkspaceParams",
     "AdminEmptyParams",
+    "CallCapabilityParams",
     "CreateDraftFromCapabilityParams",
     "CreateWrapperFromWorkspaceParams",
     "DeleteDeploymentParams",

src/wf_transport_rpc_http/client/capabilities.py

@@ -31,3 +31,19 @@ class RpcCapabilityClientMixin:
             "workflow.capabilities.inspect",
             {"qualified_name": qualified_name},
         )
+
+    async def call_capability(
+        self,
+        *,
+        qualified_name: str,
+        payload: dict[str, Any],
+        deployment_id: str | None = None,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.capabilities.call",
+            {
+                "qualified_name": qualified_name,
+                "payload": payload,
+                "deployment_id": deployment_id,
+            },
+        )

src/wf_transport_rpc_http/methods/capabilities.py

@@ -7,7 +7,11 @@ import fastapi_jsonrpc as jsonrpc
 from wf_server import WorkflowServer
 
 from ..errors import WorkflowRpcError, raise_workflow_rpc_error
-from ..models import InspectCapabilityParams, ListCapabilitiesParams
+from ..models import (
+    CallCapabilityParams,
+    InspectCapabilityParams,
+    ListCapabilitiesParams,
+)
 from ..params import RpcParams
 
 
@@ -41,3 +45,16 @@ def register_methods(
             )
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.capabilities.call", errors=[WorkflowRpcError])
+    async def workflow_capabilities_call(
+        params: CallCapabilityParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.call_capability(
+                qualified_name=params.qualified_name,
+                payload=params.payload,
+                deployment_id=params.deployment_id,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)

src/wf_transport_rpc_http/models.py

@@ -54,6 +54,12 @@ class InspectCapabilityParams(RpcParamsModel):
     qualified_name: str = Field(min_length=1)
 
 
+class CallCapabilityParams(RpcParamsModel):
+    qualified_name: str = Field(min_length=1)
+    payload: dict[str, Any] = Field(default_factory=dict)
+    deployment_id: str | None = None
+
+
 class CreateDraftFromCapabilityParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
     capability_name: str = Field(min_length=1)

tests/wf_cli/test_remote_target.py

@@ -327,6 +327,20 @@ def test_wf_cap_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
             "100",
         ],
     )
+    called = runner.invoke(
+        app,
+        [
+            "--config",
+            str(config_path),
+            "--url",
+            "http://test/rpc",
+            "cap",
+            "call",
+            "wf.std.constant",
+            "--input",
+            '{"value": "hello cap call"}',
+        ],
+    )
 
     assert inspected.exit_code == 0, inspected.output
     assert '"name": "wf.std.constant"' in inspected.output
@@ -336,6 +350,11 @@ def test_wf_cap_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
     assert {
         capability["source_id"] for capability in listed_payload["capabilities"]
     } == {"wf.std"}
+    assert called.exit_code == 0, called.output
+    called_payload = json.loads(called.output)
+    assert called_payload["qualified_name"] == "wf.std.constant"
+    assert called_payload["outcome"] == "ok"
+    assert called_payload["output"] == {"value": "hello cap call"}
 
 
 def test_wf_source_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:

tests/wf_transport_rpc_http/test_app.py

@@ -38,6 +38,14 @@ async def test_rpc_health_and_capability_methods(tmp_path) -> None:
             "workflow.capabilities.inspect",
             {"qualified_name": "wf.std.constant"},
         )
+        called = await _rpc(
+            client,
+            "workflow.capabilities.call",
+            {
+                "qualified_name": "wf.std.constant",
+                "payload": {"value": "hello direct rpc"},
+            },
+        )
 
     assert health_response.status_code == 200
     assert health_response.json()["status"] == "ok"
@@ -47,6 +55,10 @@ async def test_rpc_health_and_capability_methods(tmp_path) -> None:
         capability["source_id"] for capability in listed["result"]["capabilities"]
     } == {"wf.std"}
     assert inspected["result"]["name"] == "wf.std.constant"
+    assert called["result"]["qualified_name"] == "wf.std.constant"
+    assert called["result"]["kind"] == "node_spec"
+    assert called["result"]["outcome"] == "ok"
+    assert called["result"]["output"] == {"value": "hello direct rpc"}
 
 
 async def test_rpc_unknown_method_returns_json_rpc_error(tmp_path) -> None:

tests/wf_transport_rpc_http/test_client.py

@@ -71,12 +71,19 @@ async def test_rpc_workflow_client_lists_and_inspects_capabilities(tmp_path) ->
         )
         listed = await client.list_capabilities(source_id="wf.std", limit=5)
         inspected = await client.inspect_capability(qualified_name="wf.std.constant")
+        called = await client.call_capability(
+            qualified_name="wf.std.constant",
+            payload={"value": "hello rpc client"},
+        )
 
     assert listed["capabilities"]
     assert {capability["source_id"] for capability in listed["capabilities"]} == {
         "wf.std"
     }
     assert inspected["name"] == "wf.std.constant"
+    assert called["qualified_name"] == "wf.std.constant"
+    assert called["outcome"] == "ok"
+    assert called["output"] == {"value": "hello rpc client"}
 
 
 async def test_rpc_workflow_client_lists_and_inspects_sources(tmp_path) -> None: