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

refactor: group rpc transport client and methods

Browse Source
This commit is contained in:
lda
2026-06-08 21:44:41 +07:00 Verified
parent 7582c1d6fb
commit c39d361873
25 changed files with 104 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_transport_rpc_http/client/__init__.py
+37
View File
@@ -0,0 +1,37 @@
from __future__ import annotations
from dataclasses import dataclass
from .admin import RpcAdminClientMixin
from .artifacts import RpcArtifactClientMixin
from .base import RpcClientTransport
from .capabilities import RpcCapabilityClientMixin
from .deployments import RpcDeploymentClientMixin
from .drafts import RpcDraftClientMixin
from .runs import RpcRunClientMixin
from .source_registry import RpcSourceRegistryClientMixin
from .sources import RpcSourceAdminClientMixin
@dataclass(slots=True)
class RpcWorkflowApiClient(
RpcClientTransport,
RpcCapabilityClientMixin,
RpcDraftClientMixin,
RpcArtifactClientMixin,
RpcDeploymentClientMixin,
RpcRunClientMixin,
RpcSourceAdminClientMixin,
RpcSourceRegistryClientMixin,
RpcAdminClientMixin,
):
"""WorkflowApiSurface implementation backed by JSON-RPC HTTP calls.
The inheritance order is intentional: `RpcClientTransport` owns dataclass
fields and `_call`; domain mixins are stateless method groups that may only
depend on `_call`. If a domain needs state or lifecycle later, prefer a
composed domain client instead of adding fields to a mixin.
"""
__all__ = ["RpcWorkflowApiClient"]
src/wf_transport_rpc_http/client/admin.py
+52
View File
@@ -0,0 +1,52 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
class RpcAdminClientMixin:
"""JSON-RPC implementation of read-only admin/config surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
async def list_connections(self) -> dict[str, Any]:
return await self._call("workflow.admin.connections.list", {})
async def get_connection_statuses(self) -> dict[str, Any]:
return await self._call("workflow.admin.connection_statuses.list", {})
async def list_events(self) -> dict[str, Any]:
return await self._call("workflow.admin.events.list", {})
async def list_auth_records(self) -> dict[str, Any]:
return await self._call("workflow.admin.auth.list", {})
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
return await self._call(
"workflow.admin.auth.inspect",
{"auth_ref": auth_ref},
)
async def save_auth_record(
self,
*,
auth_ref: str,
scheme: str,
payload: Mapping[str, object],
metadata: Mapping[str, object] | None = None,
) -> dict[str, Any]:
return await self._call(
"workflow.admin.auth.save",
{
"auth_ref": auth_ref,
"scheme": scheme,
"payload": dict(payload),
"metadata": dict(metadata or {}),
},
)
async def delete_auth_record(self, auth_ref: str) -> dict[str, Any]:
return await self._call(
"workflow.admin.auth.delete",
{"auth_ref": auth_ref},
)
src/wf_transport_rpc_http/client/artifacts.py
+38
View File
@@ -0,0 +1,38 @@
from __future__ import annotations
from typing import Any, Literal
class RpcArtifactClientMixin:
"""JSON-RPC implementation of workflow artifact surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
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})
src/wf_transport_rpc_http/client/base.py
+46
View File
@@ -0,0 +1,46 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from uuid import uuid4
import httpx
@dataclass(slots=True)
class RpcClientTransport:
"""Shared JSON-RPC request plumbing for workflow RPC client mixins.
If `http_client` is provided, that client's own timeout configuration wins;
`timeout_seconds` is only used when this transport creates an `AsyncClient`.
"""
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
src/wf_transport_rpc_http/client/capabilities.py
+33
View File
@@ -0,0 +1,33 @@
from __future__ import annotations
from typing import Any
class RpcCapabilityClientMixin:
"""JSON-RPC implementation of workflow capability surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
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},
)
src/wf_transport_rpc_http/client/deployments.py
+35
View File
@@ -0,0 +1,35 @@
from __future__ import annotations
from typing import Any
class RpcDeploymentClientMixin:
"""JSON-RPC implementation of workflow deployment surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
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},
)
src/wf_transport_rpc_http/client/drafts.py
+138
View File
@@ -0,0 +1,138 @@
from __future__ import annotations
from collections.abc import Sequence
from typing import Any, Literal
class RpcDraftClientMixin:
"""JSON-RPC implementation of workflow draft workspace surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
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,
},
)
src/wf_transport_rpc_http/client/runs.py
+70
View File
@@ -0,0 +1,70 @@
from __future__ import annotations
from typing import Any
from wf_api.runs import TraceRangeLike
class RpcRunClientMixin:
"""JSON-RPC implementation of workflow run lifecycle surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
async def run_deployment(
self,
*,
deployment_id: str,
workflow_input: dict[str, Any],
trace_range: TraceRangeLike | 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 resume_run(
self,
*,
run_id: str,
resume_payload: dict[str, Any],
resume_outcome: str = "submitted",
trace_range: TraceRangeLike | None = None,
) -> dict[str, Any]:
return await self._call(
"workflow.runs.resume",
{
"run_id": run_id,
"resume_payload": resume_payload,
"resume_outcome": resume_outcome,
"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: TraceRangeLike,
) -> 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: TraceRangeLike | None,
) -> dict[str, int] | None:
if trace_range is None:
return None
return {"start": trace_range.start, "limit": trace_range.limit}
src/wf_transport_rpc_http/client/source_registry.py
+87
View File
@@ -0,0 +1,87 @@
from __future__ import annotations
from typing import Any
class RpcSourceRegistryClientMixin:
"""JSON-RPC implementation of source registry surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
async def list_registry_entries(
self,
*,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.list",
{"cursor": cursor, "limit": limit},
)
async def inspect_registry_entry(
self,
*,
source_id: str,
) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.inspect",
{"source_id": source_id},
)
async def add_registry_entry(
self,
*,
entry: dict[str, Any],
) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.add",
{"entry": entry},
)
async def update_registry_entry(
self,
*,
source_id: str,
patch: dict[str, Any],
) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.update",
{"source_id": source_id, "patch": patch},
)
async def enable_registry_entry(
self,
*,
source_id: str,
) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.enable",
{"source_id": source_id},
)
async def disable_registry_entry(
self,
*,
source_id: str,
) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.disable",
{"source_id": source_id},
)
async def remove_registry_entry(
self,
*,
source_id: str,
) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.remove",
{"source_id": source_id},
)
async def apply_registry_changes(self) -> dict[str, Any]:
return await self._call(
"workflow.admin.source_registry.apply",
{},
)
src/wf_transport_rpc_http/client/sources.py
+29
View File
@@ -0,0 +1,29 @@
from __future__ import annotations
from typing import Any
class RpcSourceAdminClientMixin:
"""JSON-RPC implementation of read-only source admin surface methods."""
async def _call(self, method: str, params: dict[str, Any]) -> dict[str, Any]: ...
async def list_sources(
self,
*,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._call(
"workflow.sources.list",
{
"cursor": cursor,
"limit": limit,
},
)
async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
return await self._call(
"workflow.sources.inspect",
{"source_id": source_id},
)