lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add remote workflow lifecycle rpc methods

Browse Source
This commit is contained in:
lda
2026-06-03 11:18:32 +07:00 Verified
parent 9155dae8c8
commit 826b28682f
12 changed files with 820 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/current_roadmap.md
+3 -3
View File
@@ -94,9 +94,9 @@ implementation state.
`WorkflowServer` through fixed dotted methods. Remote CLI targeting remains
the next transport-facing slice.
- Completed: workflow config now distinguishes client targets from server
hosting config, selected `wf cap`/`wf run` commands can target JSON-RPC
HTTP with explicit CLI overrides, and local-only draft/artifact/deploy
commands fail fast for remote targets until they are wired.
hosting config, the basic `wf` lifecycle can target JSON-RPC HTTP:
capability discovery, draft workspace authoring, artifact/deployment
operations, run, inspect, and bounded trace.
5. **CLI/API alignment**
- Let the CLI target either local process-backed stores/runtime or the future
docs/superpowers/specs/2026-06-03-workflow-config-targets-and-sources.md
+3 -3
View File
@@ -462,8 +462,9 @@ First slice implemented:
wired source ids (`wf.std`, `wf.recipes`)
- local and JSON-RPC client targets
- `wf` root overrides for `--local`, `--url`, and `--timeout`
- remote JSON-RPC client support for capability and run CLI commands
- local-only CLI commands fail fast for `rpc_http` targets until they are wired
- remote JSON-RPC client support now covers capability, draft workspace,
artifact, deployment, and run CLI commands
- draft/artifact/deploy commands no longer fail fast for `rpc_http` targets
- `wf-rpc-server --config` support for server store and RPC HTTP transport,
including configured RPC path
@@ -473,7 +474,6 @@ Still future:
- MCP/OpenAPI source config
- arbitrary stdlib source aliases
- `/mcp` hosting from neutral server config
- remote draft/artifact/deployment CLI commands
- auth and SQL stores
## Next Implementation Slice
src/wf_cli/commands/artifacts.py
+1 -1
View File
@@ -5,7 +5,7 @@ from typing import Annotated, Literal
import typer
from wf_cli.context import load_local_cli_context_from_typer as load_cli_context
from wf_cli.context import load_cli_context_from_typer as load_cli_context
from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import emit_json
src/wf_cli/commands/deployments.py
+1 -1
View File
@@ -6,7 +6,7 @@ from typing import Annotated
import typer
from wf_cli.context import load_local_cli_context_from_typer as load_cli_context
from wf_cli.context import load_cli_context_from_typer as load_cli_context
from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_input
src/wf_cli/commands/drafts.py
+1 -1
View File
@@ -6,7 +6,7 @@ from typing import Annotated, Literal
import typer
from wf_cli.context import load_local_cli_context_from_typer as load_cli_context
from wf_cli.context import load_cli_context_from_typer as load_cli_context
from wf_cli.formats import ListOutputFormat, emit_list_payload
from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_value
src/wf_transport_rpc_http/__init__.py
+22
View File
@@ -4,12 +4,22 @@ from .app import create_rpc_app
from .client import RpcWorkflowApiClient
from .errors import WorkflowRpcError
from .models import (
CreateArtifactFromWorkspaceParams,
CreateDraftFromCapabilityParams,
CreateWrapperFromWorkspaceParams,
DeleteDeploymentParams,
GetDraftWorkspaceParams,
HealthParams,
InspectArtifactParams,
InspectCapabilityParams,
InspectDeploymentParams,
InspectRunParams,
ListArtifactsParams,
ListCapabilitiesParams,
ListDeploymentsParams,
ListDraftWorkspacesParams,
PatchDraftParams,
PatchDraftWorkspaceParams,
ReadRunTraceParams,
ResumeRunParams,
SaveArtifactParams,
@@ -18,15 +28,26 @@ from .models import (
TraceRangeParams,
ValidateDeploymentParams,
ValidateDraftParams,
ValidateDraftWorkspaceParams,
)
__all__ = [
"CreateArtifactFromWorkspaceParams",
"CreateDraftFromCapabilityParams",
"CreateWrapperFromWorkspaceParams",
"DeleteDeploymentParams",
"GetDraftWorkspaceParams",
"HealthParams",
"InspectArtifactParams",
"InspectCapabilityParams",
"InspectDeploymentParams",
"InspectRunParams",
"ListArtifactsParams",
"ListCapabilitiesParams",
"ListDeploymentsParams",
"ListDraftWorkspacesParams",
"PatchDraftParams",
"PatchDraftWorkspaceParams",
"ReadRunTraceParams",
"ResumeRunParams",
"SaveArtifactParams",
@@ -35,6 +56,7 @@ __all__ = [
"TraceRangeParams",
"ValidateDeploymentParams",
"ValidateDraftParams",
"ValidateDraftWorkspaceParams",
"WorkflowRpcError",
"create_rpc_app",
"RpcWorkflowApiClient",
src/wf_transport_rpc_http/app.py
+187
View File
@@ -10,11 +10,21 @@ from wf_server import WorkflowServer
from .errors import WorkflowRpcError, raise_workflow_rpc_error
from .models import (
CreateArtifactFromWorkspaceParams,
CreateDraftFromCapabilityParams,
CreateWrapperFromWorkspaceParams,
DeleteDeploymentParams,
GetDraftWorkspaceParams,
InspectArtifactParams,
InspectCapabilityParams,
InspectDeploymentParams,
InspectRunParams,
ListArtifactsParams,
ListCapabilitiesParams,
ListDeploymentsParams,
ListDraftWorkspacesParams,
PatchDraftParams,
PatchDraftWorkspaceParams,
ReadRunTraceParams,
ResumeRunParams,
SaveArtifactParams,
@@ -22,6 +32,7 @@ from .models import (
StartRunParams,
ValidateDeploymentParams,
ValidateDraftParams,
ValidateDraftWorkspaceParams,
)
@@ -116,6 +127,125 @@ def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.draft_workspaces.list", errors=[WorkflowRpcError])
async def workflow_draft_workspaces_list(
params: ListDraftWorkspacesParams = Body(
default_factory=ListDraftWorkspacesParams
),
) -> dict[str, Any]:
try:
return await server.api.list_draft_workspaces()
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.draft_workspaces.get", errors=[WorkflowRpcError])
async def workflow_draft_workspaces_get(
params: GetDraftWorkspaceParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.get_draft_workspace(
workspace_id=params.workspace_id,
include_draft=params.include_draft,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(
name="workflow.draft_workspaces.create_from_capability",
errors=[WorkflowRpcError],
)
async def workflow_draft_workspaces_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.draft_workspaces.patch", errors=[WorkflowRpcError]
)
async def workflow_draft_workspaces_patch(
params: PatchDraftWorkspaceParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.patch_draft_workspace(
workspace_id=params.workspace_id,
revision=params.revision,
patch=params.patch,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(
name="workflow.draft_workspaces.validate", errors=[WorkflowRpcError]
)
async def workflow_draft_workspaces_validate(
params: ValidateDraftWorkspaceParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.validate_draft_workspace(
workspace_id=params.workspace_id,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(
name="workflow.draft_workspaces.create_artifact", errors=[WorkflowRpcError]
)
async def workflow_draft_workspaces_create_artifact(
params: CreateArtifactFromWorkspaceParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.create_artifact_from_workspace(
workspace_id=params.workspace_id,
artifact_id=params.artifact_id,
version=params.version,
title=params.title,
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.draft_workspaces.create_wrapper", errors=[WorkflowRpcError]
)
async def workflow_draft_workspaces_create_wrapper(
params: CreateWrapperFromWorkspaceParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.create_wrapper_from_workspace(
workspace_id=params.workspace_id,
artifact_id=params.artifact_id,
version=params.version,
title=params.title,
outcomes=tuple(params.outcomes),
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 = Params(...), # type: ignore[reportArgumentType],
@@ -146,6 +276,63 @@ def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.artifacts.list", errors=[WorkflowRpcError])
async def workflow_artifacts_list(
params: ListArtifactsParams = Body(default_factory=ListArtifactsParams),
) -> dict[str, Any]:
try:
return await server.api.list_artifacts(
query=params.query,
kind=params.kind,
cursor=params.cursor,
limit=params.limit,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.artifacts.inspect", errors=[WorkflowRpcError])
async def workflow_artifacts_inspect(
params: InspectArtifactParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.inspect_artifact(
artifact_id=params.artifact_id,
version=params.version,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.deployments.list", errors=[WorkflowRpcError])
async def workflow_deployments_list(
params: ListDeploymentsParams = Body(default_factory=ListDeploymentsParams),
) -> dict[str, Any]:
try:
return await server.api.list_deployments()
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.deployments.inspect", errors=[WorkflowRpcError])
async def workflow_deployments_inspect(
params: InspectDeploymentParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.inspect_deployment(
deployment_id=params.deployment_id,
)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
raise_workflow_rpc_error(exc)
@entrypoint.method(name="workflow.deployments.delete", errors=[WorkflowRpcError])
async def workflow_deployments_delete(
params: DeleteDeploymentParams = Params(...), # type: ignore[reportArgumentType],
) -> dict[str, Any]:
try:
return await server.api.delete_deployment(
deployment_id=params.deployment_id,
)
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],
src/wf_transport_rpc_http/client.py
+196 -2
View File
@@ -1,7 +1,8 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any
from typing import Any, Literal
from uuid import uuid4
import httpx
@@ -14,7 +15,7 @@ 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.
methods used by CLI commands targeting rpc_http transports.
"""
url: str
@@ -47,6 +48,8 @@ class RpcWorkflowApiClient:
raise RuntimeError("JSON-RPC response result must be an object")
return result
# -- capabilities --
async def list_capabilities(
self,
*,
@@ -71,6 +74,197 @@ class RpcWorkflowApiClient:
{"qualified_name": qualified_name},
)
# -- draft workspaces --
async def list_draft_workspaces(self) -> dict[str, Any]:
return await self._call("workflow.draft_workspaces.list", {})
async def get_draft_workspace(
self,
*,
workspace_id: str,
include_draft: bool = False,
) -> dict[str, Any]:
return await self._call(
"workflow.draft_workspaces.get",
{"workspace_id": workspace_id, "include_draft": include_draft},
)
async def create_draft_workspace_from_capability(
self,
*,
workspace_id: str,
capability_name: str,
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: Sequence[Any] | None = None,
output: Sequence[Any] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: Any | None = None,
) -> dict[str, Any]:
return await self._call(
"workflow.draft_workspaces.create_from_capability",
{
"workspace_id": workspace_id,
"capability_name": capability_name,
"name": name,
"title": title,
"input_schema": input_schema,
"state_schema": state_schema,
"output_schema": output_schema,
"input": input,
"output": output,
"input_map": input_map,
"output_map": output_map,
"error_message_source": error_message_source,
},
)
async def patch_draft_workspace(
self,
*,
workspace_id: str,
revision: int,
patch: list[dict[str, Any]],
) -> dict[str, Any]:
return await self._call(
"workflow.draft_workspaces.patch",
{"workspace_id": workspace_id, "revision": revision, "patch": patch},
)
async def validate_draft_workspace(
self,
*,
workspace_id: str,
) -> dict[str, Any]:
return await self._call(
"workflow.draft_workspaces.validate",
{"workspace_id": workspace_id},
)
async def create_artifact_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
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.draft_workspaces.create_artifact",
{
"workspace_id": workspace_id,
"artifact_id": artifact_id,
"version": version,
"title": title,
"outcomes": list(outcomes),
"kind": kind,
"description": description,
"required_capabilities": required_capabilities,
"source_bindings": source_bindings,
"created_from_catalog_version": created_from_catalog_version,
},
)
async def create_wrapper_from_workspace(
self,
*,
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: Sequence[str],
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.draft_workspaces.create_wrapper",
{
"workspace_id": workspace_id,
"artifact_id": artifact_id,
"version": version,
"title": title,
"outcomes": list(outcomes),
"description": description,
"required_capabilities": required_capabilities,
"source_bindings": source_bindings,
"created_from_catalog_version": created_from_catalog_version,
},
)
# -- artifacts --
async def list_artifacts(
self,
*,
query: str | None = None,
kind: Literal["workflow", "wrapper"] | None = None,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._call(
"workflow.artifacts.list",
{
"query": query,
"kind": kind,
"cursor": cursor,
"limit": limit,
},
)
async def inspect_artifact(
self, *, artifact_id: str, version: int
) -> dict[str, Any]:
return await self._call(
"workflow.artifacts.inspect",
{"artifact_id": artifact_id, "version": version},
)
async def save_artifact(self, artifact: dict[str, Any]) -> dict[str, Any]:
return await self._call("workflow.artifacts.save", {"artifact": artifact})
# -- deployments --
async def list_deployments(self) -> dict[str, Any]:
return await self._call("workflow.deployments.list", {})
async def inspect_deployment(self, *, deployment_id: str) -> dict[str, Any]:
return await self._call(
"workflow.deployments.inspect",
{"deployment_id": deployment_id},
)
async def validate_deployment(
self, *, deployment_id: str, live_check: bool = False
) -> dict[str, Any]:
return await self._call(
"workflow.deployments.validate",
{"deployment_id": deployment_id, "live_check": live_check},
)
async def save_deployment(self, deployment: dict[str, Any]) -> dict[str, Any]:
return await self._call("workflow.deployments.save", {"deployment": deployment})
async def delete_deployment(self, *, deployment_id: str) -> dict[str, Any]:
return await self._call(
"workflow.deployments.delete",
{"deployment_id": deployment_id},
)
# -- runs --
async def run_deployment(
self,
*,
src/wf_transport_rpc_http/models.py
+69 -1
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from typing import Any
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
@@ -73,6 +73,74 @@ class SaveDeploymentParams(RpcParamsModel):
deployment: dict[str, Any]
class ListDraftWorkspacesParams(RpcParamsModel):
pass
class GetDraftWorkspaceParams(RpcParamsModel):
workspace_id: str = Field(min_length=1)
include_draft: bool = False
class PatchDraftWorkspaceParams(RpcParamsModel):
workspace_id: str = Field(min_length=1)
revision: int = Field(ge=1)
patch: list[dict[str, Any]]
class ValidateDraftWorkspaceParams(RpcParamsModel):
workspace_id: str = Field(min_length=1)
class CreateArtifactFromWorkspaceParams(RpcParamsModel):
workspace_id: str = Field(min_length=1)
artifact_id: str = Field(min_length=1)
version: int = Field(ge=1)
title: str = Field(min_length=1)
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 CreateWrapperFromWorkspaceParams(RpcParamsModel):
workspace_id: str = Field(min_length=1)
artifact_id: str = Field(min_length=1)
version: int = Field(ge=1)
title: str = Field(min_length=1)
outcomes: list[str]
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
cursor: str | None = None
limit: int = Field(default=50, ge=1, le=100)
class InspectArtifactParams(RpcParamsModel):
artifact_id: str = Field(min_length=1)
version: int = Field(ge=1)
class ListDeploymentsParams(RpcParamsModel):
pass
class InspectDeploymentParams(RpcParamsModel):
deployment_id: str = Field(min_length=1)
class DeleteDeploymentParams(RpcParamsModel):
deployment_id: str = Field(min_length=1)
class ValidateDeploymentParams(RpcParamsModel):
deployment_id: str = Field(min_length=1)
live_check: bool = False
tests/wf_cli/test_remote_target.py
+94
View File
@@ -247,3 +247,97 @@ def test_wf_cap_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
)
assert result.exit_code == 0, result.output
assert '"name": "wf.std.constant"' in result.output
def test_wf_remote_draft_artifact_deploy_lifecycle(monkeypatch, tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
original_client = httpx.AsyncClient
monkeypatch.setattr(
"wf_transport_rpc_http.client.httpx.AsyncClient",
lambda *args, **kwargs: original_client(
transport=httpx.ASGITransport(app=create_rpc_app(server)),
base_url="http://test",
),
)
config_path = tmp_path / "wf.json"
config_path.write_text('{"version": 1}', encoding="utf-8")
runner = CliRunner()
base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
created = runner.invoke(
app,
[
*base_args,
"draft",
"create-from-capability",
"remote_ws",
"wf.std.constant",
"--name",
"remote_constant",
"--title",
"Remote Constant",
],
)
assert created.exit_code == 0, created.output
assert '"workspace_id": "remote_ws"' in created.output
validated = runner.invoke(
app,
[*base_args, "draft", "validate", "remote_ws"],
)
assert validated.exit_code == 0, validated.output
assert '"status": "valid"' in validated.output
saved_artifact = runner.invoke(
app,
[
*base_args,
"draft",
"save",
"remote_ws",
"--artifact",
"remote_artifact",
"--version",
"1",
"--title",
"Remote Artifact",
"--outcome",
"ok",
"--binding",
"wf.std=wf.std",
],
)
assert saved_artifact.exit_code == 0, saved_artifact.output
assert '"artifact_id": "remote_artifact"' in saved_artifact.output
inspected_artifact = runner.invoke(
app,
[*base_args, "artifact", "inspect", "remote_artifact", "1"],
)
assert inspected_artifact.exit_code == 0, inspected_artifact.output
assert '"id": "remote_artifact"' in inspected_artifact.output
saved_deployment = runner.invoke(
app,
[
*base_args,
"deploy",
"save",
"remote_artifact.default",
"--artifact",
"remote_artifact",
"--version",
"1",
"--binding",
"wf.std=wf.std",
],
)
assert saved_deployment.exit_code == 0, saved_deployment.output
assert '"deployment_id": "remote_artifact.default"' in saved_deployment.output
validated_deployment = runner.invoke(
app,
[*base_args, "deploy", "validate", "remote_artifact.default"],
)
assert validated_deployment.exit_code == 0, validated_deployment.output
assert '"status": "runnable"' in validated_deployment.output
tests/wf_transport_rpc_http/test_app.py
+118
View File
@@ -211,6 +211,124 @@ def test_rpc_draft_artifact_deployment_lifecycle(tmp_path) -> None:
asyncio.run(scenario())
def test_rpc_artifact_and_deployment_catalog_methods(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_lifecycle",
version=1,
title="RPC Lifecycle",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
)
await server.api.save_deployment(
{
"id": "rpc_lifecycle.default",
"artifact_id": "rpc_lifecycle",
"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:
listed_artifacts = await _rpc(client, "workflow.artifacts.list", {})
inspected_artifact = await _rpc(
client,
"workflow.artifacts.inspect",
{"artifact_id": "rpc_lifecycle", "version": 1},
)
listed_deployments = await _rpc(client, "workflow.deployments.list", {})
inspected_deployment = await _rpc(
client,
"workflow.deployments.inspect",
{"deployment_id": "rpc_lifecycle.default"},
)
deleted = await _rpc(
client,
"workflow.deployments.delete",
{"deployment_id": "rpc_lifecycle.default"},
)
assert listed_artifacts["result"]["nodes"]
assert inspected_artifact["result"]["id"] == "rpc_lifecycle"
assert listed_deployments["result"]["deployments"]
assert inspected_deployment["result"]["id"] == "rpc_lifecycle.default"
assert deleted["result"]["deployment_id"] == "rpc_lifecycle.default"
asyncio.run(scenario())
def test_rpc_draft_workspace_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:
created = await _rpc(
client,
"workflow.draft_workspaces.create_from_capability",
{
"workspace_id": "remote_ws",
"capability_name": "wf.std.constant",
"name": "remote_constant",
"title": "Remote Constant",
"input_map": {},
"output_map": {"value": "state.result"},
},
)
listed = await _rpc(client, "workflow.draft_workspaces.list", {})
fetched = await _rpc(
client,
"workflow.draft_workspaces.get",
{"workspace_id": "remote_ws"},
)
validated = await _rpc(
client,
"workflow.draft_workspaces.validate",
{"workspace_id": "remote_ws"},
)
patched = await _rpc(
client,
"workflow.draft_workspaces.patch",
{
"workspace_id": "remote_ws",
"revision": created["result"]["revision"],
"patch": [
{"op": "replace", "path": "/name", "value": "remote_renamed"}
],
},
)
artifact = await _rpc(
client,
"workflow.draft_workspaces.create_artifact",
{
"workspace_id": "remote_ws",
"artifact_id": "remote_artifact",
"version": 1,
"title": "Remote Artifact",
"outcomes": ["ok"],
"kind": "workflow",
"source_bindings": {"wf.std": "wf.std"},
},
)
assert created["result"]["workspace_id"] == "remote_ws"
assert listed["result"]["workspaces"]
assert fetched["result"]["workspace_id"] == "remote_ws"
assert validated["result"]["status"] in {"valid", "invalid"}
assert patched["result"]["revision"] == created["result"]["revision"] + 1
assert artifact["result"]["artifact_id"] == "remote_artifact"
asyncio.run(scenario())
def _constant_plan() -> RawWorkflowPlan:
return RawWorkflowPlan.model_validate(
{
tests/wf_transport_rpc_http/test_client.py
+125
View File
@@ -157,3 +157,128 @@ def test_rpc_workflow_client_raises_for_rpc_error(tmp_path) -> None:
assert "missing.capability" in message
asyncio.run(scenario())
def test_rpc_workflow_client_lists_and_inspects_artifacts(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_art",
version=1,
title="Client Art",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "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
)
listed = await client.list_artifacts()
inspected = await client.inspect_artifact(
artifact_id="client_art", version=1
)
assert listed["nodes"]
assert inspected["id"] == "client_art"
asyncio.run(scenario())
def test_rpc_workflow_client_lists_inspects_validates_and_deletes_deployments(
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_deploy_art",
version=1,
title="Client Deploy Art",
plan=_constant_plan(),
outcomes=["ok"],
source_bindings={"wf.std": "wf.std"},
)
await server.api.save_deployment(
{
"id": "client_deploy_art.default",
"artifact_id": "client_deploy_art",
"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
)
listed = await client.list_deployments()
inspected = await client.inspect_deployment(
deployment_id="client_deploy_art.default"
)
validated = await client.validate_deployment(
deployment_id="client_deploy_art.default"
)
deleted = await client.delete_deployment(
deployment_id="client_deploy_art.default"
)
assert listed["deployments"]
assert inspected["id"] == "client_deploy_art.default"
assert validated["status"] == "runnable"
assert deleted["deployment_id"] == "client_deploy_art.default"
asyncio.run(scenario())
def test_rpc_workflow_client_draft_workspace_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 http_client:
client = RpcWorkflowApiClient(
url="http://test/rpc", timeout_seconds=5, http_client=http_client
)
created = await client.create_draft_workspace_from_capability(
workspace_id="client_ws",
capability_name="wf.std.constant",
name="client_constant",
title="Client Constant",
input_map={},
output_map={"value": "state.result"},
)
listed = await client.list_draft_workspaces()
fetched = await client.get_draft_workspace(workspace_id="client_ws")
validated = await client.validate_draft_workspace(workspace_id="client_ws")
patched = await client.patch_draft_workspace(
workspace_id="client_ws",
revision=created["revision"],
patch=[{"op": "replace", "path": "/name", "value": "client_renamed"}],
)
artifact = await client.create_artifact_from_workspace(
workspace_id="client_ws",
artifact_id="client_ws_art",
version=1,
title="Client WS Art",
outcomes=("ok",),
kind="workflow",
source_bindings={"wf.std": "wf.std"},
)
assert created["workspace_id"] == "client_ws"
assert listed["workspaces"]
assert fetched["workspace_id"] == "client_ws"
assert validated["status"] in {"valid", "invalid"}
assert patched["revision"] == created["revision"] + 1
assert artifact["artifact_id"] == "client_ws_art"
asyncio.run(scenario())