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

workflow builder deprecated field stop use in wf_mcp

Browse Source
This commit is contained in:
lda
2026-05-21 05:32:40 +07:00 Verified
parent d089e28c15
commit 8df4c94c4e
13 changed files with 463 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_artifacts/drafts/adapter.py
+25 -3
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
from typing import Any
from wf_authoring import WorkflowBuilder
from wf_authoring.dsl import PathExpr
from wf_core import JoinNode, Workflow
@@ -42,9 +44,8 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
return builder.use_ref(
step.use,
id=step_id,
in_map=step.in_,
input_values=step.with_,
out_map=step.out,
input=_draft_input_bindings(step),
output=_draft_output_bindings(step),
desc=step.desc,
)
if isinstance(step, DraftForeachStep):
@@ -88,3 +89,24 @@ def _add_step(builder: WorkflowBuilder, step_id: str, step: DraftStep):
default=step.match.default,
).entry
raise TypeError(f"unsupported draft step {type(step)!r}")
def _draft_input_bindings(step: DraftUseStep) -> list[dict[str, Any]]:
"""Translate draft input maps into canonical core input binding structs.
Draft JSON keeps `in` and `with` because they are compact patch targets for
LLM clients. The compiled workflow should not re-emit deprecated builder map
sugar, so this adapter boundary converts them to `NodeUse.input`.
"""
literal_bindings = [
{"target": target, "value": value} for target, value in step.with_.items()
]
path_bindings = [
{"target": target, "path": source} for source, target in step.in_.items()
]
return [*literal_bindings, *path_bindings]
def _draft_output_bindings(step: DraftUseStep) -> list[dict[str, str]]:
"""Translate draft output maps into canonical core output binding structs."""
return [{"source": source, "target": target} for source, target in step.out.items()]
src/wf_mcp/workflow_surface/handlers.py
+113 -44
View File
@@ -32,6 +32,12 @@ from wf_platform import (
)
from wf_authoring import build_async_registry
from wf_core import RuntimeContext
from wf_core.models.steps import (
InputBinding,
InputPathBinding,
InputValueBinding,
OutputBinding,
)
from ..events import make_event
from ..models import RawWorkflowPlan
@@ -214,19 +220,21 @@ class WorkflowSurfaceHandlers:
query=query,
):
continue
rows.append({
"name": name,
"source_id": "workflow",
"kind": "wrapper_artifact",
"artifact_id": artifact.id,
"version": artifact.version,
"title": artifact.title,
"description": artifact.description,
"outcomes": list(artifact.outcomes),
"is_async": True,
"input_fields": _schema_field_names(artifact.input_schema),
"output_fields": _schema_field_names(artifact.output_schema),
})
rows.append(
{
"name": name,
"source_id": "workflow",
"kind": "wrapper_artifact",
"artifact_id": artifact.id,
"version": artifact.version,
"title": artifact.title,
"description": artifact.description,
"outcomes": list(artifact.outcomes),
"is_async": True,
"input_fields": _schema_field_names(artifact.input_schema),
"output_fields": _schema_field_names(artifact.output_schema),
}
)
return rows
def _wrapper_capability_detail(
@@ -445,10 +453,12 @@ class WorkflowSurfaceHandlers:
},
)
)
required_sources = sorted({
capability.logical_source
for capability in workflow_artifact.required_capability_map().values()
})
required_sources = sorted(
{
capability.logical_source
for capability in workflow_artifact.required_capability_map().values()
}
)
return {
"artifact_id": workflow_artifact.id,
"version": workflow_artifact.version,
@@ -628,26 +638,34 @@ class WorkflowSurfaceHandlers:
input_schema: dict[str, Any],
state_schema: dict[str, Any],
output_schema: dict[str, Any],
input_map: dict[str, str],
output_map: dict[str, str],
input: Sequence[InputBinding] | None = None,
output: Sequence[OutputBinding] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: str | None = None,
title: str | None = None,
) -> dict[str, Any]:
"""Bootstrap the smallest patchable draft around one workflow capability."""
draft_input, draft_with = _draft_input_maps(
input=input,
input_map=input_map,
)
draft_output = _draft_output_map(output=output, output_map=output_map)
outcomes = self._outcomes_for_capability(capability_name) or (
DEFAULT_OK_OUTCOME,
)
steps: dict[str, Any] = {
DEFAULT_CALL_STEP_ID: {
"use": capability_name,
"in": input_map,
"out": output_map,
"in": draft_input,
"with": draft_with,
"out": draft_output,
}
}
routes: dict[str, dict[str, str]] = {
DEFAULT_CALL_STEP_ID: {DEFAULT_OK_OUTCOME: "__end__"}
}
error_source = error_message_source or _first_state_path(output_map)
error_source = error_message_source or _first_state_path(draft_output)
if DEFAULT_ERROR_OUTCOME in outcomes and error_source is not None:
# The bootstrapper cannot infer provider-specific error envelopes.
# It only wires an error route when the caller gave, or output_map
@@ -684,6 +702,8 @@ class WorkflowSurfaceHandlers:
input_schema: dict[str, Any] | None = None,
state_schema: dict[str, Any] | None = None,
output_schema: dict[str, Any] | None = None,
input: Sequence[InputBinding] | None = None,
output: Sequence[OutputBinding] | None = None,
input_map: dict[str, str] | None = None,
output_map: dict[str, str] | None = None,
error_message_source: str | None = None,
@@ -698,8 +718,12 @@ class WorkflowSurfaceHandlers:
input_schema=input_schema or hints["input_schema"],
state_schema=state_schema or hints["state_schema"],
output_schema=output_schema or hints["output_schema"],
input_map=input_map or hints["input_map"],
output_map=output_map or hints["output_map"],
input=input,
output=output,
input_map=None if input is not None else (input_map or hints["input_map"]),
output_map=None
if output is not None
else (output_map or hints["output_map"]),
error_message_source=error_message_source,
title=title,
)
@@ -905,14 +929,16 @@ def _available_sources(service: WfMcpService) -> list[AvailableSource]:
if (capability_name := _capability_name(spec.name)) is not None
if (detail := node_spec_details.get(spec.name)) is not None
}
capabilities.update({
capability_name: AvailableCapability(
name=capability_name,
kind="reducer",
)
for reducer in source.capabilities.reducers.values()
if (capability_name := _capability_name(reducer.name)) is not None
})
capabilities.update(
{
capability_name: AvailableCapability(
name=capability_name,
kind="reducer",
)
for reducer in source.capabilities.reducers.values()
if (capability_name := _capability_name(reducer.name)) is not None
}
)
sources.append(
AvailableSource(
id=source.id,
@@ -976,9 +1002,9 @@ def _observed_node_specs(service: WfMcpService) -> dict[str, NodeSpecInventory]:
observed: dict[str, NodeSpecInventory] = {}
for source in service.capability_sources.values():
inventory = source.as_inventory()
observed.update({
detail.name: detail for detail in inventory.capabilities.node_spec_details
})
observed.update(
{detail.name: detail for detail in inventory.capabilities.node_spec_details}
)
return observed
@@ -1013,6 +1039,47 @@ def _first_state_path(output_map: dict[str, str]) -> str | None:
return None
def _draft_input_maps(
*,
input: Sequence[InputBinding] | None,
input_map: dict[str, str] | None,
) -> tuple[dict[str, str], dict[str, Any]]:
"""Convert canonical MCP input bindings into draft `in` and `with` maps.
Draft workspaces intentionally keep compact maps as patch targets, while
MCP-facing request models prefer the canonical core binding structs. This
helper keeps that translation explicit at the frontend boundary.
"""
if input is not None and input_map is not None:
raise ValueError("cannot mix canonical input bindings with input_map")
if input is None:
return dict(input_map or {}), {}
mapped_inputs: dict[str, str] = {}
literal_inputs: dict[str, Any] = {}
for binding in input:
if isinstance(binding, InputPathBinding):
mapped_inputs[str(binding.path)] = str(binding.target)
elif isinstance(binding, InputValueBinding):
literal_inputs[str(binding.target)] = binding.value
else: # pragma: no cover - defensive against future input binding variants.
raise TypeError(f"unsupported input binding {binding!r}")
return mapped_inputs, literal_inputs
def _draft_output_map(
*,
output: Sequence[OutputBinding] | None,
output_map: dict[str, str] | None,
) -> dict[str, str]:
"""Convert canonical MCP output bindings into the draft `out` map."""
if output is not None and output_map is not None:
raise ValueError("cannot mix canonical output bindings with output_map")
if output is None:
return dict(output_map or {})
return {str(binding.source): str(binding.target) for binding in output}
def _escape_json_pointer(value: str) -> str:
"""Escape one JSON Pointer path segment for generated JSON Patch helpers."""
return value.replace("~", "~0").replace("/", "~1")
@@ -1057,15 +1124,17 @@ def _artifact_capability_id(artifact: WorkflowArtifact) -> str:
def _raw_plan_from_artifact(artifact: WorkflowArtifact) -> RawWorkflowPlan:
"""Validate the stored plan shape expected by the broker workflow runner."""
return RawWorkflowPlan.model_validate({
"name": _plan_field(artifact, "name"),
"input_schema": _plan_field(artifact, "input_schema"),
"state_schema": _plan_field(artifact, "state_schema"),
"output_schema": _plan_field(artifact, "output_schema"),
"start": _plan_field(artifact, "start"),
"nodes": _plan_field(artifact, "nodes"),
"edges": _plan_field(artifact, "edges"),
})
return RawWorkflowPlan.model_validate(
{
"name": _plan_field(artifact, "name"),
"input_schema": _plan_field(artifact, "input_schema"),
"state_schema": _plan_field(artifact, "state_schema"),
"output_schema": _plan_field(artifact, "output_schema"),
"start": _plan_field(artifact, "start"),
"nodes": _plan_field(artifact, "nodes"),
"edges": _plan_field(artifact, "edges"),
}
)
def _plan_field(artifact: WorkflowArtifact, field_name: str) -> Any:
src/wf_mcp/workflow_surface/models.py
+62 -15
View File
@@ -6,6 +6,7 @@ from pydantic import BaseModel, Field
from wf_artifacts import ArtifactKind
from wf_artifacts.draft_workspaces.models import WORKSPACE_ID_PATTERN
from wf_core.models.steps import InputBinding, OutputBinding
WorkspaceId = Annotated[
str,
@@ -27,8 +28,29 @@ DraftPathMap = Annotated[
dict[str, str],
Field(
description=(
"Map local draft paths to workflow paths, for example "
"{'input.text': 'text'} or {'echoed': 'state.echoed'}."
"Compatibility map form for draft paths. Prefer canonical input/output "
"binding lists for new MCP/JSON clients."
)
),
]
DraftInputBindings = Annotated[
list[InputBinding],
Field(
description=(
"Canonical node input bindings. Use path bindings such as "
"{'target': {'root': 'local', 'parts': ['text']}, "
"'path': {'root': 'input', 'parts': ['text']}} or value bindings "
"with {'target': ..., 'value': ...}."
)
),
]
DraftOutputBindings = Annotated[
list[OutputBinding],
Field(
description=(
"Canonical node output bindings. Example: {'source': {'root': "
"'local', 'parts': ['echoed']}, 'target': {'root': 'state', "
"'parts': ['echoed']}}."
)
),
]
@@ -203,25 +225,40 @@ class CreateMinimalDraftWorkspaceRequest(BaseModel):
)
state_schema: JsonSchemaObject = Field(
description=(
"Workflow state schema. The current core state schema uses a fields "
"object; keep it small and explicit."
"Workflow state JSON Schema. Prefer object properties with reducer "
"extension keywords; legacy fields input is compatibility-only."
)
)
output_schema: JsonSchemaObject = Field(
description="Public output JSON Schema for the workflow or wrapper being drafted."
)
input_map: DraftPathMap = Field(
input: DraftInputBindings | None = Field(
default=None,
description=(
"Map public workflow input paths to local capability input paths. "
"Example: {'input.text': 'message'} sends workflow input.text to "
"capability field message."
)
"Preferred canonical input bindings for the called capability. "
"Use this instead of input_map for new clients."
),
)
output_map: DraftPathMap = Field(
output: DraftOutputBindings | None = Field(
default=None,
description=(
"Map local capability output paths to workflow state paths. Example: "
"{'echoed': 'state.echoed'} stores capability output echoed in state.echoed."
)
"Preferred canonical output bindings for writes into workflow state. "
"Use this instead of output_map for new clients."
),
)
input_map: DraftPathMap | None = Field(
default=None,
description=(
"Deprecated compatibility map from workflow paths to local capability "
"input paths, for example {'input.text': 'message'}."
),
)
output_map: DraftPathMap | None = Field(
default=None,
description=(
"Deprecated compatibility map from local capability output paths to "
"workflow state paths, for example {'echoed': 'state.echoed'}."
),
)
error_message_source: str | None = Field(
default=None,
@@ -260,13 +297,23 @@ class CreateDraftWorkspaceFromCapabilityRequest(BaseModel):
default=None,
description="Optional override for the hinted public output schema.",
)
input: DraftInputBindings | None = Field(
default=None,
description="Optional canonical override for the hinted workflow input bindings.",
)
output: DraftOutputBindings | None = Field(
default=None,
description="Optional canonical override for the hinted capability output bindings.",
)
input_map: DraftPathMap | None = Field(
default=None,
description="Optional override for the hinted workflow input map.",
description="Deprecated compatibility override for the hinted workflow input map.",
)
output_map: DraftPathMap | None = Field(
default=None,
description="Optional override for the hinted capability output map.",
description=(
"Deprecated compatibility override for the hinted capability output map."
),
)
error_message_source: str | None = Field(
default=None,
src/wf_mcp/workflow_surface/tools.py
+14 -3
View File
@@ -340,7 +340,10 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
@server.tool(
name="wf.workflow.set_step_input_map",
title="Set Step Input Map",
description="Replace one step input map in a draft workspace.",
description=(
"Replace one compatibility step input map in a draft workspace. "
"New one-capability bootstraps should prefer canonical input bindings."
),
)
async def set_step_input_map(
request: SetStepInputMapRequest,
@@ -357,7 +360,10 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
@server.tool(
name="wf.workflow.set_step_output_map",
title="Set Step Output Map",
description="Replace one step output map in a draft workspace.",
description=(
"Replace one compatibility step output map in a draft workspace. "
"New one-capability bootstraps should prefer canonical output bindings."
),
)
async def set_step_output_map(
request: SetStepOutputMapRequest,
@@ -376,7 +382,8 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
title="Create Minimal Draft Workspace",
description=(
"Bootstrap a patchable draft workspace around one inspected capability. "
"Use this before patch helpers when authoring from MCP clients."
"Use canonical input/output binding lists for new MCP clients; "
"input_map/output_map remain compatibility fields."
),
)
async def create_minimal_draft_workspace(
@@ -390,6 +397,8 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
input_schema=request.input_schema,
state_schema=request.state_schema,
output_schema=request.output_schema,
input=request.input,
output=request.output,
input_map=request.input_map,
output_map=request.output_map,
error_message_source=request.error_message_source,
@@ -417,6 +426,8 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
input_schema=request.input_schema,
state_schema=request.state_schema,
output_schema=request.output_schema,
input=request.input,
output=request.output,
input_map=request.input_map,
output_map=request.output_map,
error_message_source=request.error_message_source,
src/wf_platform/sources.py
+3
View File
@@ -6,6 +6,7 @@ from pydantic import BaseModel
from wf_authoring import NodeSpec
from wf_core import ReducerSpec
from wf_core.runtime.ops.merges import ReducerDefinition
from wf_platform.refs import CapabilityRef
SourceKind = Literal["system", "connection"]
JsonObject = dict[str, Any]
@@ -59,6 +60,7 @@ class ReducerInventory(BaseModel):
"""Serializable public contract for one pure reducer."""
name: str
ref: CapabilityRef
description: str | None = None
config_schema: JsonObject
@@ -200,6 +202,7 @@ class CapabilitySource:
reducer_details=tuple(
ReducerInventory(
name=reducer.name,
ref=CapabilityRef.parse(reducer.name),
description=reducer.description,
config_schema=reducer.config_schema,
)