feat: expose create artifact from plan over rpc

src/wf_transport_rpc_http/__init__.py

@@ -6,6 +6,7 @@ from .errors import WorkflowRpcError
 from .models import (
     AdminEmptyParams,
     CallCapabilityParams,
+    CreateArtifactFromPlanParams,
     CreateArtifactFromWorkspaceParams,
     CreateDraftFromCapabilityParams,
     CreateWrapperFromWorkspaceParams,
@@ -36,6 +37,7 @@ from .models import (
 )
 
 __all__ = [
+    "CreateArtifactFromPlanParams",
     "CreateArtifactFromWorkspaceParams",
     "AdminEmptyParams",
     "CallCapabilityParams",

src/wf_transport_rpc_http/client/artifacts.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Sequence
 from typing import Any, Literal
 
 from .base import RpcCaller
@@ -46,3 +47,33 @@ class RpcArtifactClientMixin:
             "workflow.artifacts.delete",
             {"artifact_id": artifact_id, "version": version},
         )
+
+    async def create_artifact_from_plan(
+        self: RpcCaller,
+        *,
+        artifact_id: str,
+        version: int,
+        title: str,
+        plan: dict[str, Any],
+        outcomes: Sequence[str],
+        kind: Literal["workflow", "wrapper"] = "workflow",
+        description: str | None = None,
+        required_capabilities: dict[str, dict[str, Any]] | None = None,
+        source_bindings: dict[str, str] | None = None,
+        created_from_catalog_version: str | None = None,
+    ) -> dict[str, Any]:
+        return await self._call(
+            "workflow.artifacts.create_from_plan",
+            {
+                "artifact_id": artifact_id,
+                "version": version,
+                "title": title,
+                "plan": plan,
+                "outcomes": list(outcomes),
+                "kind": kind,
+                "description": description,
+                "required_capabilities": required_capabilities,
+                "source_bindings": source_bindings,
+                "created_from_catalog_version": created_from_catalog_version,
+            },
+        )

src/wf_transport_rpc_http/methods/artifacts.py

@@ -8,6 +8,7 @@ from wf_server import WorkflowServer
 
 from ..errors import WorkflowRpcError, raise_workflow_rpc_error
 from ..models import (
+    CreateArtifactFromPlanParams,
     DeleteArtifactParams,
     InspectArtifactParams,
     ListArtifactsParams,
@@ -22,6 +23,29 @@ def register_methods(
 ) -> None:
     """Register artifact JSON-RPC methods."""
 
+    @entrypoint.method(
+        name="workflow.artifacts.create_from_plan",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_artifacts_create_from_plan(
+        params: CreateArtifactFromPlanParams = RpcParams(),
+    ) -> dict[str, Any]:
+        try:
+            return await server.api.create_artifact_from_plan(
+                artifact_id=params.artifact_id,
+                version=params.version,
+                title=params.title,
+                plan=params.plan,
+                outcomes=tuple(params.outcomes),
+                kind=params.kind,
+                description=params.description,
+                required_capabilities=params.required_capabilities,
+                source_bindings=params.source_bindings,
+                created_from_catalog_version=params.created_from_catalog_version,
+            )
+        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 = RpcParams(),

src/wf_transport_rpc_http/models.py

@@ -144,6 +144,19 @@ class CreateWrapperFromWorkspaceParams(RpcParamsModel):
     created_from_catalog_version: str | None = None
 
 
+class CreateArtifactFromPlanParams(RpcParamsModel):
+    artifact_id: str = Field(min_length=1)
+    version: int = Field(ge=1)
+    title: str = Field(min_length=1)
+    plan: dict[str, Any]
+    outcomes: list[str]
+    kind: Literal["workflow", "wrapper"] = "workflow"
+    description: str | None = None
+    required_capabilities: dict[str, dict[str, Any]] | None = None
+    source_bindings: dict[str, str] | None = None
+    created_from_catalog_version: str | None = None
+
+
 class ListArtifactsParams(RpcParamsModel):
     query: str | None = None
     kind: Literal["workflow", "wrapper"] | None = None

tests/wf_transport_rpc_http/test_app.py

@@ -615,6 +615,35 @@ async def test_rpc_runs_workflow_from_python_source_capability(tmp_path) -> None
     assert run["result"]["output"] == {"echoed": "hello workflow"}
 
 
+async def test_rpc_create_artifact_from_plan(tmp_path) -> 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:
+        created = await _rpc(
+            client,
+            "workflow.artifacts.create_from_plan",
+            {
+                "artifact_id": "rpc_plan",
+                "version": 1,
+                "title": "RPC Plan",
+                "plan": _constant_plan().model_dump(mode="json", by_alias=True),
+                "outcomes": ["ok"],
+                "source_bindings": {},
+            },
+        )
+        inspected = await _rpc(
+            client,
+            "workflow.artifacts.inspect",
+            {"artifact_id": "rpc_plan", "version": 1},
+        )
+
+    assert created["result"]["artifact_id"] == "rpc_plan"
+    assert created["result"]["version"] == 1
+    assert inspected["result"]["id"] == "rpc_plan"
+    assert inspected["result"]["plan"]["name"] == "rpc_constant"
+
+
 async def test_rpc_diagnoses_source(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     app = create_rpc_app(server)

tests/wf_transport_rpc_http/test_client.py

@@ -415,6 +415,36 @@ async def test_rpc_client_lists_runs(tmp_path) -> None:
     assert listed["runs"][0]["run_id"] == started["run_id"]
 
 
+async def test_rpc_client_creates_artifact_from_plan(tmp_path) -> 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,
+        )
+        created = await client.create_artifact_from_plan(
+            artifact_id="client_plan",
+            version=1,
+            title="Client Plan",
+            plan=_constant_plan().model_dump(mode="json", by_alias=True),
+            outcomes=("ok",),
+            source_bindings={},
+        )
+        inspected = await client.inspect_artifact(
+            artifact_id="client_plan",
+            version=1,
+        )
+
+    assert created["artifact_id"] == "client_plan"
+    assert inspected["id"] == "client_plan"
+
+
 async def test_rpc_client_diagnoses_source(tmp_path) -> None:
     calls: list[tuple[str, dict[str, object]]] = []