feat: expose workflow capabilities over json rpc

src/wf_transport_rpc_http/__init__.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from .app import create_rpc_app
+from .errors import WorkflowRpcError
 from .models import (
     CreateDraftFromCapabilityParams,
     HealthParams,
@@ -32,4 +34,6 @@ __all__ = [
     "TraceRangeParams",
     "ValidateDeploymentParams",
     "ValidateDraftParams",
+    "WorkflowRpcError",
+    "create_rpc_app",
 ]

src/wf_transport_rpc_http/app.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from typing import Any
+
+import fastapi_jsonrpc as jsonrpc
+from fastapi import Body
+from fastapi_jsonrpc import Params
+
+from wf_server import WorkflowServer
+
+from .errors import WorkflowRpcError, raise_workflow_rpc_error
+from .models import InspectCapabilityParams, ListCapabilitiesParams
+
+
+def create_rpc_app(server: WorkflowServer) -> jsonrpc.API:
+    """Build a JSON-RPC HTTP app over an existing WorkflowServer.
+
+    Transport code owns only JSON-RPC envelope handling. Workflow semantics stay
+    behind server.api, so this package remains swappable with WebSocket/MCP
+    transports later.
+    """
+
+    app = jsonrpc.API()
+    entrypoint = jsonrpc.Entrypoint("/rpc")
+
+    @app.get("/healthz")
+    async def healthz() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @entrypoint.method(name="workflow.health", errors=[WorkflowRpcError])
+    async def workflow_health() -> dict[str, Any]:
+        return {
+            "status": "ok",
+            "store_root": str(server.config.store_root),
+        }
+
+    @entrypoint.method(name="workflow.capabilities.list", errors=[WorkflowRpcError])
+    async def workflow_capabilities_list(
+        params: ListCapabilitiesParams = Body(default_factory=ListCapabilitiesParams),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.list_capabilities(
+                query=params.query,
+                source_id=params.source_id,
+                cursor=params.cursor,
+                limit=params.limit,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.capabilities.inspect", errors=[WorkflowRpcError])
+    async def workflow_capabilities_inspect(
+        params: InspectCapabilityParams = Params(...),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.inspect_capability(
+                qualified_name=params.qualified_name,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    app.bind_entrypoint(entrypoint)
+    return app

src/wf_transport_rpc_http/errors.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from typing import NoReturn
+
+import fastapi_jsonrpc as jsonrpc
+from pydantic import BaseModel, ConfigDict
+
+
+class WorkflowRpcError(jsonrpc.BaseError):
+    """Expected workflow application error surfaced through JSON-RPC."""
+
+    CODE = 5000
+    MESSAGE = "Workflow operation failed"
+
+    class DataModel(BaseModel):
+        code: str
+        message: str
+
+        model_config = ConfigDict(extra="forbid")
+
+
+def raise_workflow_rpc_error(exc: Exception) -> NoReturn:
+    """Map expected application exceptions without swallowing programming bugs."""
+
+    raise WorkflowRpcError(
+        data={
+            "code": exc.__class__.__name__,
+            "message": str(exc),
+        }
+    ) from exc

tests/wf_transport_rpc_http/test_app.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import httpx
+
+from wf_server import build_local_static_workflow_server
+from wf_transport_rpc_http.app import create_rpc_app
+
+
+async def _rpc(client: httpx.AsyncClient, method: str, params: dict[str, Any]) -> dict[str, Any]:
+    response = await client.post(
+        "/rpc",
+        json={"jsonrpc": "2.0", "id": "test", "method": method, "params": params},
+    )
+    assert response.status_code == 200
+    return response.json()
+
+
+def test_rpc_health_and_capability_methods(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 client:
+            health_response = await client.get("/healthz")
+            health = await _rpc(client, "workflow.health", {})
+            listed = await _rpc(
+                client,
+                "workflow.capabilities.list",
+                {"source_id": "wf.std", "limit": 10},
+            )
+            inspected = await _rpc(
+                client,
+                "workflow.capabilities.inspect",
+                {"qualified_name": "wf.std.constant"},
+            )
+
+        assert health_response.status_code == 200
+        assert health_response.json()["status"] == "ok"
+        assert health["result"]["status"] == "ok"
+        assert listed["result"]["capabilities"]
+        assert inspected["result"]["name"] == "wf.std.constant"
+
+    asyncio.run(scenario())
+
+
+def test_rpc_unknown_method_returns_json_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 client:
+            payload = await _rpc(client, "workflow.nope", {})
+
+        assert payload["error"]["code"] == -32601
+        assert payload["error"]["message"] == "Method not found"
+
+    asyncio.run(scenario())