feat: add workflow rpc http client

src/wf_transport_rpc_http/__init__.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from .app import create_rpc_app
+from .client import RpcWorkflowApiClient
 from .errors import WorkflowRpcError
 from .models import (
     CreateDraftFromCapabilityParams,
@@ -36,4 +37,5 @@ __all__ = [
     "ValidateDraftParams",
     "WorkflowRpcError",
     "create_rpc_app",
+    "RpcWorkflowApiClient",
 ]

src/wf_transport_rpc_http/client.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+from uuid import uuid4
+
+import httpx
+
+from wf_api.models import TraceRange
+
+
+@dataclass(slots=True)
+class RpcWorkflowApiClient:
+    """Small WorkflowApi-compatible adapter for JSON-RPC HTTP targets.
+
+    This is intentionally not a full WorkflowApi clone. It implements only the
+    methods used by the first remote CLI slice.
+    """
+
+    url: str
+    timeout_seconds: float = 30.0
+    http_client: httpx.AsyncClient | None = None
+
+    async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
+        request = {
+            "jsonrpc": "2.0",
+            "id": uuid4().hex,
+            "method": method,
+            "params": params,
+        }
+        if self.http_client is None:
+            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+                response = await client.post(self.url, json=request)
+        else:
+            response = await self.http_client.post(self.url, json=request)
+        response.raise_for_status()
+        payload = response.json()
+        if "error" in payload:
+            error = payload["error"]
+            message = error.get("message", "JSON-RPC error")
+            data = error.get("data")
+            if isinstance(data, dict) and data.get("message"):
+                message = f"{message}: {data['message']}"
+            raise RuntimeError(message)
+        result = payload.get("result")
+        if not isinstance(result, dict):
+            raise RuntimeError("JSON-RPC response result must be an object")
+        return result
+
+    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 await self._call(
+            "workflow.capabilities.list",
+            {
+                "query": query,
+                "source_id": source_id,
+                "cursor": cursor,
+                "limit": limit,
+            },
+        )
+
+    async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
+        return await self._call(
+            "workflow.capabilities.inspect",
+            {"qualified_name": qualified_name},
+        )
+
+    async def run_deployment(
+        self,
+        *,
+        deployment_id: str,
+        workflow_input: dict[str, Any],
+        trace_range: TraceRange | None = None,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.runs.start",
+            {
+                "deployment_id": deployment_id,
+                "workflow_input": workflow_input,
+                "trace_range": _trace_range_payload(trace_range),
+            },
+        )
+
+    async def inspect_run(self, *, run_id: str) -> dict[str, Any]:
+        return await self._call("workflow.runs.inspect", {"run_id": run_id})
+
+    async def read_run_trace(
+        self,
+        *,
+        run_id: str,
+        trace_range: TraceRange,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.runs.trace",
+            {
+                "run_id": run_id,
+                "trace_range": _trace_range_payload(trace_range),
+            },
+        )
+
+
+def _trace_range_payload(trace_range: TraceRange | None) -> dict[str, int] | None:
+    if trace_range is None:
+        return None
+    return {"start": trace_range.start, "limit": trace_range.limit}

tests/wf_transport_rpc_http/test_client.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import httpx
+
+from wf_api.models import RawWorkflowPlan, TraceRange
+from wf_core import END
+from wf_server import build_local_static_workflow_server
+from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
+
+
+def _constant_plan() -> RawWorkflowPlan:
+    return RawWorkflowPlan.model_validate(
+        {
+            "name": "client_constant",
+            "input_schema": {"type": "object", "properties": {}},
+            "state_schema": {
+                "type": "object",
+                "properties": {
+                    "result": {"type": "string", "reducer": "wf.std.replace"}
+                },
+            },
+            "output_schema": {
+                "type": "object",
+                "properties": {"result": {"type": "string"}},
+                "required": ["result"],
+            },
+            "outcomes": ["ok"],
+            "start": "constant",
+            "nodes": [
+                {
+                    "id": "constant",
+                    "type": "node",
+                    "node": "wf.std.constant",
+                    "input": [
+                        {
+                            "value": "hello from rpc client",
+                            "target": {"root": "local", "parts": ["value"]},
+                        }
+                    ],
+                    "output": [
+                        {
+                            "source": {"root": "local", "parts": ["value"]},
+                            "target": {"root": "state", "parts": ["result"]},
+                        }
+                    ],
+                }
+            ],
+            "edges": [{"from": "constant", "outcome": "ok", "to": END}],
+            "output": [
+                {
+                    "path": {"root": "state", "parts": ["result"]},
+                    "target": {"root": "local", "parts": ["result"]},
+                }
+            ],
+        }
+    )
+
+
+def test_rpc_workflow_client_lists_and_inspects_capabilities(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport,
+            base_url="http://test",
+        ) as http_client:
+            client = RpcWorkflowApiClient(
+                url="http://test/rpc",
+                timeout_seconds=5,
+                http_client=http_client,
+            )
+            listed = await client.list_capabilities(source_id="wf.std", limit=5)
+            inspected = await client.inspect_capability(
+                qualified_name="wf.std.constant"
+            )
+
+        assert listed["capabilities"]
+        assert inspected["name"] == "wf.std.constant"
+
+    asyncio.run(scenario())
+
+
+def test_rpc_workflow_client_runs_and_reads_trace(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        await server.api.create_artifact_from_plan(
+            artifact_id="client_constant",
+            version=1,
+            title="Client Constant",
+            plan=_constant_plan(),
+            outcomes=["ok"],
+            source_bindings={"wf.std": "wf.std"},
+        )
+        await server.api.save_deployment(
+            {
+                "id": "client_constant.default",
+                "artifact_id": "client_constant",
+                "artifact_version": 1,
+                "bindings": [{"logical_source": "wf.std", "concrete_source": "wf.std"}],
+            }
+        )
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport,
+            base_url="http://test",
+        ) as http_client:
+            client = RpcWorkflowApiClient(
+                url="http://test/rpc",
+                timeout_seconds=5,
+                http_client=http_client,
+            )
+            run = await client.run_deployment(
+                deployment_id="client_constant.default",
+                workflow_input={},
+                trace_range=TraceRange(start=0, limit=1),
+            )
+            inspected = await client.inspect_run(run_id=run["run_id"])
+            trace = await client.read_run_trace(
+                run_id=run["run_id"],
+                trace_range=TraceRange(start=0, limit=1),
+            )
+
+        assert run["status"] == "completed"
+        assert run["output"]["result"] == "hello from rpc client"
+        assert inspected["trace_count"] >= 1
+        assert len(trace["trace"]) == 1
+
+    asyncio.run(scenario())
+
+
+def test_rpc_workflow_client_raises_for_rpc_error(tmp_path) -> None:
+    async def scenario() -> None:
+        server = build_local_static_workflow_server(tmp_path / "store")
+        app = create_rpc_app(server)
+        transport = httpx.ASGITransport(app=app)
+        async with httpx.AsyncClient(
+            transport=transport,
+            base_url="http://test",
+        ) as http_client:
+            client = RpcWorkflowApiClient(
+                url="http://test/rpc",
+                timeout_seconds=5,
+                http_client=http_client,
+            )
+            try:
+                await client.inspect_capability(qualified_name="missing.capability")
+            except RuntimeError as exc:
+                message = str(exc)
+            else:
+                raise AssertionError("expected RuntimeError")
+
+        assert "Workflow operation failed" in message
+        assert "missing.capability" in message
+
+    asyncio.run(scenario())