feat: expose input bindings over rpc
This commit is contained in:
@@ -43,6 +43,7 @@ from .models import (
|
||||
SetDraftNameParams,
|
||||
SetDraftRouteParams,
|
||||
SetDraftStartParams,
|
||||
SetStepInputBindingsParams,
|
||||
SetStepInputMapParams,
|
||||
SetStepOutputMapParams,
|
||||
SetWorkflowOutputMapParams,
|
||||
@@ -93,6 +94,7 @@ __all__ = [
|
||||
"SetDraftNameParams",
|
||||
"SetDraftRouteParams",
|
||||
"SetDraftStartParams",
|
||||
"SetStepInputBindingsParams",
|
||||
"SetStepInputMapParams",
|
||||
"SetStepOutputMapParams",
|
||||
"SetWorkflowOutputMapParams",
|
||||
|
||||
@@ -5,6 +5,7 @@ from typing import Any, Literal
|
||||
|
||||
from wf_api.surface import RouteSource
|
||||
from wf_artifacts.drafts.models import DraftStep
|
||||
from wf_core.models.steps import InputBinding
|
||||
|
||||
from .base import RpcCaller
|
||||
|
||||
@@ -182,6 +183,24 @@ class RpcDraftClientMixin:
|
||||
},
|
||||
)
|
||||
|
||||
async def set_step_input_bindings(
|
||||
self: RpcCaller,
|
||||
*,
|
||||
workspace_id: str,
|
||||
revision: int,
|
||||
step_id: str,
|
||||
bindings: Sequence[InputBinding],
|
||||
) -> dict[str, Any]:
|
||||
return await self._call(
|
||||
"workflow.draft_workspaces.set_step_input_bindings",
|
||||
{
|
||||
"workspace_id": workspace_id,
|
||||
"revision": revision,
|
||||
"step_id": step_id,
|
||||
"bindings": [binding.model_dump(mode="json") for binding in bindings],
|
||||
},
|
||||
)
|
||||
|
||||
async def set_step_output_map(
|
||||
self: RpcCaller,
|
||||
*,
|
||||
|
||||
@@ -31,6 +31,7 @@ from ..models import (
|
||||
SetDraftNameParams,
|
||||
SetDraftRouteParams,
|
||||
SetDraftStartParams,
|
||||
SetStepInputBindingsParams,
|
||||
SetStepInputMapParams,
|
||||
SetStepOutputMapParams,
|
||||
SetWorkflowOutputMapParams,
|
||||
@@ -226,6 +227,23 @@ def register_methods(
|
||||
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
|
||||
raise_workflow_rpc_error(exc)
|
||||
|
||||
@entrypoint.method(
|
||||
name="workflow.draft_workspaces.set_step_input_bindings",
|
||||
errors=[WorkflowRpcError],
|
||||
)
|
||||
async def workflow_draft_workspaces_set_step_input_bindings(
|
||||
params: SetStepInputBindingsParams = RpcParams(),
|
||||
) -> dict[str, Any]:
|
||||
try:
|
||||
return await server.api.set_step_input_bindings(
|
||||
workspace_id=params.workspace_id,
|
||||
revision=params.revision,
|
||||
step_id=params.step_id,
|
||||
bindings=params.bindings,
|
||||
)
|
||||
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
|
||||
raise_workflow_rpc_error(exc)
|
||||
|
||||
@entrypoint.method(
|
||||
name="workflow.draft_workspaces.set_step_output_map",
|
||||
errors=[WorkflowRpcError],
|
||||
|
||||
@@ -6,6 +6,7 @@ from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||||
|
||||
from wf_api.models import TraceRange
|
||||
from wf_artifacts.drafts.models import DraftStep
|
||||
from wf_core.models.steps import InputBinding
|
||||
|
||||
|
||||
class RpcParamsModel(BaseModel):
|
||||
@@ -208,6 +209,13 @@ class SetStepInputMapParams(RpcParamsModel):
|
||||
merge: bool = False
|
||||
|
||||
|
||||
class SetStepInputBindingsParams(RpcParamsModel):
|
||||
workspace_id: str = Field(min_length=1)
|
||||
revision: int = Field(ge=1)
|
||||
step_id: str = Field(min_length=1)
|
||||
bindings: list[InputBinding]
|
||||
|
||||
|
||||
class SetStepOutputMapParams(RpcParamsModel):
|
||||
workspace_id: str = Field(min_length=1)
|
||||
revision: int = Field(ge=1)
|
||||
|
||||
@@ -890,6 +890,75 @@ async def test_rpc_draft_workspace_focused_edit_methods(tmp_path) -> None:
|
||||
)
|
||||
|
||||
|
||||
async def test_rpc_set_step_input_bindings_preserves_canonical_order(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:
|
||||
await _rpc(
|
||||
client,
|
||||
"workflow.draft_workspaces.create_from_capability",
|
||||
{
|
||||
"workspace_id": "binding_ws",
|
||||
"capability_name": "wf.std.concat",
|
||||
"name": "binding_transport",
|
||||
},
|
||||
)
|
||||
result = await _rpc(
|
||||
client,
|
||||
"workflow.draft_workspaces.set_step_input_bindings",
|
||||
{
|
||||
"workspace_id": "binding_ws",
|
||||
"revision": 1,
|
||||
"step_id": "call",
|
||||
"bindings": [
|
||||
{"path": "input.items", "target": "items"},
|
||||
{"value": "\n", "target": "separator"},
|
||||
],
|
||||
},
|
||||
)
|
||||
inspected = await _rpc(
|
||||
client,
|
||||
"workflow.draft_workspaces.get",
|
||||
{"workspace_id": "binding_ws", "include_draft": True},
|
||||
)
|
||||
|
||||
assert result["result"]["revision"] == 2
|
||||
assert inspected["result"]["draft"]["steps"]["call"]["input"] == [
|
||||
{"target": "items", "path": "input.items"},
|
||||
{"target": "separator", "value": "\n"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"binding",
|
||||
[
|
||||
{"path": "input.items", "value": [], "target": "items"},
|
||||
{"target": "items"},
|
||||
],
|
||||
)
|
||||
async def test_rpc_set_step_input_bindings_rejects_malformed_union(
|
||||
tmp_path,
|
||||
binding: dict[str, Any],
|
||||
) -> 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:
|
||||
rejected = await _rpc(
|
||||
client,
|
||||
"workflow.draft_workspaces.set_step_input_bindings",
|
||||
{
|
||||
"workspace_id": "missing_ws",
|
||||
"revision": 1,
|
||||
"step_id": "call",
|
||||
"bindings": [binding],
|
||||
},
|
||||
)
|
||||
|
||||
assert rejected["error"]["code"] == -32602
|
||||
|
||||
|
||||
async def test_rpc_draft_workspace_set_workflow_output_map(tmp_path) -> None:
|
||||
server = build_local_static_workflow_server(tmp_path / "store")
|
||||
app = create_rpc_app(server)
|
||||
|
||||
@@ -15,6 +15,7 @@ from wf_artifacts.drafts.models import (
|
||||
DraftStep,
|
||||
)
|
||||
from wf_core import END
|
||||
from wf_core.models.steps import InputPathBinding, InputValueBinding
|
||||
from wf_server import build_local_static_workflow_server
|
||||
from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
|
||||
from wf_transport_rpc_http.client.drafts import RpcDraftClientMixin
|
||||
@@ -689,6 +690,40 @@ async def test_rpc_client_draft_workspace_focused_edit_methods(tmp_path) -> None
|
||||
assert state_bound["revision"] == 8
|
||||
|
||||
|
||||
async def test_rpc_client_serializes_canonical_step_input_bindings() -> None:
|
||||
calls: list[tuple[str, dict[str, object]]] = []
|
||||
|
||||
class Client(RpcDraftClientMixin):
|
||||
async def _call(self, method: str, params: dict[str, object]):
|
||||
calls.append((method, params))
|
||||
return {"revision": 3}
|
||||
|
||||
client = Client()
|
||||
|
||||
await client.set_step_input_bindings(
|
||||
workspace_id="client_ws",
|
||||
revision=2,
|
||||
step_id="call",
|
||||
bindings=[
|
||||
InputPathBinding(path="state.title", target="request.title"),
|
||||
InputValueBinding(target="request.format", value="markdown"),
|
||||
],
|
||||
)
|
||||
|
||||
assert calls[-1] == (
|
||||
"workflow.draft_workspaces.set_step_input_bindings",
|
||||
{
|
||||
"workspace_id": "client_ws",
|
||||
"revision": 2,
|
||||
"step_id": "call",
|
||||
"bindings": [
|
||||
{"target": "request.title", "path": "state.title"},
|
||||
{"target": "request.format", "value": "markdown"},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
async def test_rpc_client_draft_remove_methods(tmp_path) -> None:
|
||||
calls: list[dict[str, Any]] = []
|
||||
|
||||
|
||||
Reference in New Issue
Block a user