merge json rpc workflow transport

2026-06-03 07:34:33 +07:00
parent f13f099949 5ac8f9f591
commit 24fdaadfe0
14 changed files with 2173 additions and 2 deletions
@@ -90,6 +90,9 @@ implementation state.
     transactional storage, and live upstream MCP sources.
    - First slice implemented: `wf_server` can construct a local/static durable
      `WorkflowApi` without `WfMcpService`. Transport adapters remain future work.
+    - Completed: the first JSON-RPC-over-HTTP transport can expose the local/static
+      `WorkflowServer` through fixed dotted methods. Remote CLI targeting remains
+      the next transport-facing slice.

 5. **CLI/API alignment**
   - Let the CLI target either local process-backed stores/runtime or the future
@@ -275,6 +275,15 @@ composition:

 It should not implement source provider management yet.

+Implementation status:
+
+- `wf_transport_rpc_http.create_rpc_app(server)` exposes a fixed JSON-RPC
+  method set over an existing `wf_server.WorkflowServer`.
+- `wf-rpc-server --store-root <path>` starts the local/static server over
+  `/rpc`.
+- This slice still does not include remote `wf` CLI targeting, auth,
+  streaming/progress, or live upstream MCP source management.
+
 Preferred implementation dependency:

 ```bash
@@ -6,6 +6,7 @@ readme = "readme.md"
 authors = [{ name = "lda", email = "[email protected]" }]
 requires-python = ">=3.14"
 dependencies = [
+    "fastapi-jsonrpc>=3.5.0",
    "fastmcp>=3.2.4",
    "httpx>=0.28",
    "jsonpatch>=1.33",
@@ -14,11 +15,13 @@ dependencies = [
    "openapi-core>=0.19",
    "pydantic>=2",
    "typer>=0.24.2",
+    "uvicorn>=0.46.0",
 ]

 [project.scripts]
 wf = "wf_cli.app:main"
 wf-mcp = "wf_mcp.cli:main"
+wf-rpc-server = "wf_transport_rpc_http.cli:main"

 [dependency-groups]
 dev = [
@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from .app import create_rpc_app
+from .errors import WorkflowRpcError
+from .models import (
+    CreateDraftFromCapabilityParams,
+    HealthParams,
+    InspectCapabilityParams,
+    InspectRunParams,
+    ListCapabilitiesParams,
+    PatchDraftParams,
+    ReadRunTraceParams,
+    ResumeRunParams,
+    SaveArtifactParams,
+    SaveDeploymentParams,
+    StartRunParams,
+    TraceRangeParams,
+    ValidateDeploymentParams,
+    ValidateDraftParams,
+)
+
+__all__ = [
+    "CreateDraftFromCapabilityParams",
+    "HealthParams",
+    "InspectCapabilityParams",
+    "InspectRunParams",
+    "ListCapabilitiesParams",
+    "PatchDraftParams",
+    "ReadRunTraceParams",
+    "ResumeRunParams",
+    "SaveArtifactParams",
+    "SaveDeploymentParams",
+    "StartRunParams",
+    "TraceRangeParams",
+    "ValidateDeploymentParams",
+    "ValidateDraftParams",
+    "WorkflowRpcError",
+    "create_rpc_app",
+]
@@ -0,0 +1,204 @@
+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 (
+    CreateDraftFromCapabilityParams,
+    InspectCapabilityParams,
+    InspectRunParams,
+    ListCapabilitiesParams,
+    PatchDraftParams,
+    ReadRunTraceParams,
+    ResumeRunParams,
+    SaveArtifactParams,
+    SaveDeploymentParams,
+    StartRunParams,
+    ValidateDeploymentParams,
+    ValidateDraftParams,
+)
+
+
+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(...),  # type: ignore[reportArgumentType]
+    ) -> 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)
+
+    @entrypoint.method(
+        name="workflow.drafts.create_from_capability", errors=[WorkflowRpcError]
+    )
+    async def workflow_drafts_create_from_capability(
+        params: CreateDraftFromCapabilityParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.create_draft_workspace_from_capability(
+                workspace_id=params.workspace_id,
+                capability_name=params.capability_name,
+                name=params.name,
+                title=params.title,
+                input_schema=params.input_schema,
+                state_schema=params.state_schema,
+                output_schema=params.output_schema,
+                input=params.input,
+                output=params.output,
+                input_map=params.input_map,
+                output_map=params.output_map,
+                error_message_source=params.error_message_source,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.drafts.patch", errors=[WorkflowRpcError])
+    async def workflow_drafts_patch(
+        params: PatchDraftParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.patch_draft(draft=params.draft, patch=params.patch)
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.drafts.validate", errors=[WorkflowRpcError])
+    async def workflow_drafts_validate(
+        params: ValidateDraftParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.validate_draft(draft=params.draft)
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.artifacts.save", errors=[WorkflowRpcError])
+    async def workflow_artifacts_save(
+        params: SaveArtifactParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.save_artifact(params.artifact)
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.deployments.save", errors=[WorkflowRpcError])
+    async def workflow_deployments_save(
+        params: SaveDeploymentParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.save_deployment(params.deployment)
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.deployments.validate", errors=[WorkflowRpcError])
+    async def workflow_deployments_validate(
+        params: ValidateDeploymentParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.validate_deployment(
+                deployment_id=params.deployment_id,
+                live_check=params.live_check,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.runs.start", errors=[WorkflowRpcError])
+    async def workflow_runs_start(
+        params: StartRunParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.run_deployment(
+                deployment_id=params.deployment_id,
+                workflow_input=params.workflow_input,
+                trace_range=(
+                    params.trace_range.to_api_trace_range()
+                    if params.trace_range is not None
+                    else None
+                ),
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.runs.inspect", errors=[WorkflowRpcError])
+    async def workflow_runs_inspect(
+        params: InspectRunParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.inspect_run(run_id=params.run_id)
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.runs.trace", errors=[WorkflowRpcError])
+    async def workflow_runs_trace(
+        params: ReadRunTraceParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.read_run_trace(
+                run_id=params.run_id,
+                trace_range=params.trace_range.to_api_trace_range(),
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    @entrypoint.method(name="workflow.runs.resume", errors=[WorkflowRpcError])
+    async def workflow_runs_resume(
+        params: ResumeRunParams = Params(...),  # type: ignore[reportArgumentType],
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.resume_run(
+                run_id=params.run_id,
+                resume_payload=params.resume_payload,
+                resume_outcome=params.resume_outcome,
+                trace_range=(
+                    params.trace_range.to_api_trace_range()
+                    if params.trace_range is not None
+                    else None
+                ),
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
+    app.bind_entrypoint(entrypoint)
+    return app
@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import typer
+import uvicorn
+
+from wf_server import build_local_static_workflow_server
+
+from .app import create_rpc_app
+
+app = typer.Typer(add_completion=False)
+
+
+@app.callback(invoke_without_command=True)
+def serve(
+    store_root: Path = typer.Option(
+        ...,
+        "--store-root",
+        help="Directory containing workflow artifact, draft, and run stores.",
+    ),
+    host: str = typer.Option("127.0.0.1", "--host"),
+    port: int = typer.Option(8765, "--port", min=1, max=65535),
+) -> None:
+    """Serve the local/static WorkflowApi over JSON-RPC HTTP."""
+    server = build_local_static_workflow_server(store_root)
+    rpc_app = create_rpc_app(server)
+    uvicorn.run(rpc_app, host=host, port=port, access_log=False)
+
+
+def main() -> None:
+    app()
@@ -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
@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from wf_api.models import TraceRange
+
+
+class RpcParamsModel(BaseModel):
+    """Base transport DTO: reject misspelled JSON-RPC params early."""
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class TraceRangeParams(RpcParamsModel):
+    start: int = Field(default=0, ge=0, description="Zero-based trace offset.")
+    limit: int = Field(
+        default=20,
+        ge=1,
+        le=100,
+        description="Maximum trace entries to return; full traces are never implicit.",
+    )
+
+    def to_api_trace_range(self) -> TraceRange:
+        return TraceRange(start=self.start, limit=self.limit)
+
+
+class HealthParams(RpcParamsModel):
+    pass
+
+
+class ListCapabilitiesParams(RpcParamsModel):
+    query: str | None = Field(default=None)
+    source_id: str | None = Field(default=None)
+    cursor: str | None = Field(default=None)
+    limit: int = Field(default=50, ge=1, le=200)
+
+
+class InspectCapabilityParams(RpcParamsModel):
+    qualified_name: str = Field(min_length=1)
+
+
+class CreateDraftFromCapabilityParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    capability_name: str = Field(min_length=1)
+    name: str | None = None
+    title: str | None = None
+    input_schema: dict[str, Any] | None = None
+    state_schema: dict[str, Any] | None = None
+    output_schema: dict[str, Any] | None = None
+    input: list[Any] | None = None
+    output: list[Any] | None = None
+    input_map: dict[str, str] | None = None
+    output_map: dict[str, str] | None = None
+    error_message_source: Any | None = None
+
+
+class PatchDraftParams(RpcParamsModel):
+    draft: dict[str, Any]
+    patch: list[dict[str, Any]]
+
+
+class ValidateDraftParams(RpcParamsModel):
+    draft: dict[str, Any]
+
+
+class SaveArtifactParams(RpcParamsModel):
+    artifact: dict[str, Any]
+
+
+class SaveDeploymentParams(RpcParamsModel):
+    deployment: dict[str, Any]
+
+
+class ValidateDeploymentParams(RpcParamsModel):
+    deployment_id: str = Field(min_length=1)
+    live_check: bool = False
+
+
+class StartRunParams(RpcParamsModel):
+    deployment_id: str = Field(min_length=1)
+    workflow_input: dict[str, Any] = Field(default_factory=dict)
+    trace_range: TraceRangeParams | None = None
+
+
+class InspectRunParams(RpcParamsModel):
+    run_id: str = Field(min_length=1)
+
+
+class ReadRunTraceParams(RpcParamsModel):
+    run_id: str = Field(min_length=1)
+    trace_range: TraceRangeParams
+
+
+class ResumeRunParams(RpcParamsModel):
+    run_id: str = Field(min_length=1)
+    resume_payload: dict[str, Any] = Field(default_factory=dict)
+    resume_outcome: str = Field(default="submitted", min_length=1)
+    trace_range: TraceRangeParams | None = None
@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import httpx
+
+from wf_api.models import RawWorkflowPlan
+from wf_core import END
+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())
+
+
+def test_rpc_draft_artifact_deployment_lifecycle(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:
+            draft_ws = await _rpc(
+                client,
+                "workflow.drafts.create_from_capability",
+                {
+                    "workspace_id": "constant_ws",
+                    "capability_name": "wf.std.constant",
+                    "name": "constant_workflow",
+                    "title": "Constant Workflow",
+                    "input_map": {},
+                    "output_map": {"value": "state.result"},
+                },
+            )
+
+            draft = {
+                "name": "rpc_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"],
+                },
+                "start": "constant",
+                "steps": {
+                    "constant": {
+                        "use": "wf.std.constant",
+                        "input": [
+                            {
+                                "value": "hello over rpc",
+                                "target": {"root": "local", "parts": ["value"]},
+                            }
+                        ],
+                        "output": [
+                            {
+                                "source": {"root": "local", "parts": ["value"]},
+                                "target": {"root": "state", "parts": ["result"]},
+                            }
+                        ],
+                    }
+                },
+                "routes": {"constant": {"ok": "__end__"}},
+                "output": [
+                    {
+                        "path": {"root": "state", "parts": ["result"]},
+                        "target": {"root": "local", "parts": ["result"]},
+                    }
+                ],
+            }
+
+            validate_draft = await _rpc(
+                client,
+                "workflow.drafts.validate",
+                {"draft": draft},
+            )
+            compiled_plan = validate_draft["result"]["compiled_plan"]
+            artifact = await _rpc(
+                client,
+                "workflow.artifacts.save",
+                {
+                    "artifact": {
+                        "id": "constant_rpc",
+                        "version": 1,
+                        "kind": "wrapper",
+                        "title": "Constant RPC",
+                        "input_schema": {"type": "object", "properties": {}},
+                        "output_schema": {
+                            "type": "object",
+                            "properties": {"result": {"type": "string"}},
+                            "required": ["result"],
+                        },
+                        "outcomes": ["ok"],
+                        "required_capabilities": {},
+                        "source_bindings": {"wf.std": "wf.std"},
+                        "plan": compiled_plan,
+                    },
+                },
+            )
+            deployment = await _rpc(
+                client,
+                "workflow.deployments.save",
+                {
+                    "deployment": {
+                        "id": "constant_rpc.default",
+                        "artifact_id": "constant_rpc",
+                        "artifact_version": 1,
+                        "bindings": [
+                            {"logical_source": "wf.std", "concrete_source": "wf.std"}
+                        ],
+                    },
+                },
+            )
+            validate_deployment = await _rpc(
+                client,
+                "workflow.deployments.validate",
+                {"deployment_id": "constant_rpc.default"},
+            )
+
+        assert draft_ws["result"]["workspace_id"] == "constant_ws"
+        assert validate_draft["result"]["status"] == "valid"
+        assert artifact["result"]["artifact_id"] == "constant_rpc"
+        assert deployment["result"]["deployment_id"] == "constant_rpc.default"
+        assert validate_deployment["result"]["status"] == "runnable"
+
+    asyncio.run(scenario())
+
+
+def _constant_plan() -> RawWorkflowPlan:
+    return RawWorkflowPlan.model_validate(
+        {
+            "name": "rpc_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 over rpc",
+                            "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_runs_deployment_and_reads_bounded_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="rpc_constant",
+            version=1,
+            title="RPC Constant",
+            plan=_constant_plan(),
+            outcomes=["ok"],
+            source_bindings={"wf.std": "wf.std"},
+        )
+        await server.api.save_deployment(
+            {
+                "id": "rpc_constant.default",
+                "artifact_id": "rpc_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 client:
+            run = await _rpc(
+                client,
+                "workflow.runs.start",
+                {
+                    "deployment_id": "rpc_constant.default",
+                    "workflow_input": {},
+                    "trace_range": {"start": 0, "limit": 1},
+                },
+            )
+            inspected = await _rpc(
+                client,
+                "workflow.runs.inspect",
+                {"run_id": run["result"]["run_id"]},
+            )
+            trace = await _rpc(
+                client,
+                "workflow.runs.trace",
+                {
+                    "run_id": run["result"]["run_id"],
+                    "trace_range": {"start": 0, "limit": 1},
+                },
+            )
+
+        assert run["result"]["status"] == "completed"
+        assert run["result"]["output"]["result"] == "hello over rpc"
+        assert "trace" not in inspected["result"]
+        assert inspected["result"]["trace_count"] >= 1
+        assert trace["result"]["trace_start"] == 0
+        assert trace["result"]["trace_limit"] == 1
+        assert len(trace["result"]["trace"]) == 1
+
+    asyncio.run(scenario())
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from typer.testing import CliRunner
+
+from wf_transport_rpc_http.cli import app
+
+
+def test_rpc_server_cli_help_mentions_store_root() -> None:
+    result = CliRunner().invoke(app, ["--help"])
+
+    assert result.exit_code == 0
+    assert "--store-root" in result.output
+    assert "--host" in result.output
+    assert "--port" in result.output
@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+
+def test_wf_transport_rpc_http_imports_no_wfmcp_modules() -> None:
+    root = Path("src/wf_transport_rpc_http")
+    violations: list[str] = []
+
+    for path in root.rglob("*.py"):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                if node.module == "wf_mcp" or node.module.startswith("wf_mcp."):
+                    violations.append(f"{path}:{node.lineno}: from {node.module}")
+            elif isinstance(node, ast.Import):
+                for alias in node.names:
+                    if alias.name == "wf_mcp" or alias.name.startswith("wf_mcp."):
+                        violations.append(f"{path}:{node.lineno}: import {alias.name}")
+
+    assert violations == []
@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from wf_transport_rpc_http.models import (
+    InspectCapabilityParams,
+    ListCapabilitiesParams,
+    ReadRunTraceParams,
+    StartRunParams,
+    TraceRangeParams,
+)
+
+
+def test_trace_range_params_converts_to_api_trace_range() -> None:
+    trace_range = TraceRangeParams(start=2, limit=5).to_api_trace_range()
+
+    assert trace_range.start == 2
+    assert trace_range.limit == 5
+
+
+def test_trace_range_params_rejects_invalid_values() -> None:
+    with pytest.raises(ValidationError):
+        TraceRangeParams(start=-1, limit=5)
+
+    with pytest.raises(ValidationError):
+        TraceRangeParams(start=0, limit=0)
+
+    with pytest.raises(ValidationError):
+        TraceRangeParams(start=0, limit=101)
+
+
+def test_capability_params_are_explicit_models() -> None:
+    listed = ListCapabilitiesParams(query="echo", source_id="wf.std", limit=10)
+    inspected = InspectCapabilityParams(qualified_name="wf.std.constant")
+
+    assert listed.query == "echo"
+    assert listed.source_id == "wf.std"
+    assert listed.limit == 10
+    assert inspected.qualified_name == "wf.std.constant"
+
+
+def test_run_params_are_explicit_models() -> None:
+    started = StartRunParams(
+        deployment_id="demo.default",
+        workflow_input={"message": "hello"},
+        trace_range=TraceRangeParams(start=0, limit=3),
+    )
+    trace = ReadRunTraceParams(
+        run_id="run_demo",
+        trace_range=TraceRangeParams(start=0, limit=1),
+    )
+
+    assert started.deployment_id == "demo.default"
+    assert started.workflow_input["message"] == "hello"
+    assert started.trace_range is not None
+    assert trace.run_id == "run_demo"
+    assert trace.trace_range.limit == 1
@@ -14,6 +14,15 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/50/25/da1f0b4dd970e52bf5a36c204c107e11a0c6d3ed195eba0bfbc664c312b2/aiofile-3.9.0-py3-none-any.whl", hash = "sha256:ce2f6c1571538cbdfa0143b04e16b208ecb0e9cb4148e528af8a640ed51cc8aa", size = 19539, upload-time = "2024-10-08T10:39:32.955Z" },
 ]

+[[package]]
+name = "aiojobs"
+version = "1.4.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/03/54/751969398e2039b4dc458fa153dc066a0f7337a5b480d58944f59b7b38ae/aiojobs-1.4.0.tar.gz", hash = "sha256:463665c75d1fcc46c78d44375c1034abf5e3f087894b0fc5ec4dd16ef90fdc98", size = 139598, upload-time = "2025-04-05T00:39:07.771Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ce/89/5a30b3c041712a8b2dadb5ccef2bae12a874469650c61c018705039699cd/aiojobs-1.4.0-py3-none-any.whl", hash = "sha256:e95eb0d10d1f6095aefb04d228d2b4b3747514503f26540de722c2228a8500ca", size = 9455, upload-time = "2025-04-05T00:39:06.279Z" },
+]
+
 [[package]]
 name = "annotated-doc"
 version = "0.0.4"
@@ -277,6 +286,37 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/8a/0e/97c33bf5009bdbac74fd2beace167cab3f978feb69cc36f1ef79360d6c4e/exceptiongroup-1.3.1-py3-none-any.whl", hash = "sha256:a7a39a3bd276781e98394987d3a5701d0c4edffb633bb7a5144577f82c773598", size = 16740, upload-time = "2025-11-21T23:01:53.443Z" },
 ]

+[[package]]
+name = "fastapi"
+version = "0.136.3"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "annotated-doc" },
+    { name = "pydantic" },
+    { name = "starlette" },
+    { name = "typing-extensions" },
+    { name = "typing-inspection" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/81/2d/ff8d91d7b564d464629a0fd50a4489c97fcb836ac230bf3a7269232a9b1f/fastapi-0.136.3.tar.gz", hash = "sha256:e487fae93ad408e6f47641ee4dfe389864fd7bec92e547ea8498fc13f43e83ab", size = 396410, upload-time = "2026-05-23T18:53:15.192Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e0/82/45359b62a067409bd929ae8a56b8ed13e5a8c8a61194b3c236920999ab83/fastapi-0.136.3-py3-none-any.whl", hash = "sha256:3d2a69bdf04b7e9f3afa292c3bc7a98816bbfafa10bc9b45f3f3700d2f761620", size = 117481, upload-time = "2026-05-23T18:53:16.924Z" },
+]
+
+[[package]]
+name = "fastapi-jsonrpc"
+version = "3.5.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "aiojobs" },
+    { name = "fastapi" },
+    { name = "pydantic" },
+    { name = "starlette" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/3e/c9/4e5a5eb24f7d99aa1cf1f37a2c64e9c6d79c40c19a96541345157da362c7/fastapi_jsonrpc-3.5.0.tar.gz", hash = "sha256:e4fb3a1d8e4a32aed980dad819992b34299e247a35a8d28d930f6e972f9eb1dd", size = 446763, upload-time = "2026-04-17T15:49:50.781Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/13/90/39821f8171569469852502c868a47e91c091852230db4250e89c7e8c6f46/fastapi_jsonrpc-3.5.0-py3-none-any.whl", hash = "sha256:f13b7aa6c1f020222a9e997686bd92c43e123b3cf690f743357950d4aa4201b8", size = 23802, upload-time = "2026-04-17T15:49:33.348Z" },
+]
+
 [[package]]
 name = "fastmcp"
 version = "3.3.0"
@@ -592,6 +632,7 @@ name = "lda-wf"
 version = "0.0.1"
 source = { editable = "." }
 dependencies = [
+    { name = "fastapi-jsonrpc" },
    { name = "fastmcp" },
    { name = "httpx" },
    { name = "jsonpatch" },
@@ -600,6 +641,7 @@ dependencies = [
    { name = "openapi-core" },
    { name = "pydantic" },
    { name = "typer" },
+    { name = "uvicorn" },
 ]

 [package.dev-dependencies]
@@ -609,6 +651,7 @@ dev = [

 [package.metadata]
 requires-dist = [
+    { name = "fastapi-jsonrpc", specifier = ">=3.5.0" },
    { name = "fastmcp", specifier = ">=3.2.4" },
    { name = "httpx", specifier = ">=0.28" },
    { name = "jsonpatch", specifier = ">=1.33" },
@@ -617,6 +660,7 @@ requires-dist = [
    { name = "openapi-core", specifier = ">=0.19" },
    { name = "pydantic", specifier = ">=2" },
    { name = "typer", specifier = ">=0.24.2" },
+    { name = "uvicorn", specifier = ">=0.46.0" },
 ]

 [package.metadata.requires-dev]