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

concrete typing from pydantic

Browse Source
This commit is contained in:
lda
2026-05-19 05:35:38 +07:00 Verified
parent dd577d9731
commit 1d6a70431b
8 changed files with 290 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_mcp/workflow_surface/constants.py
+15
View File
@@ -0,0 +1,15 @@
"""Shared workflow-surface literals used by generated draft helpers."""
DEFAULT_CALL_STEP_ID = "call"
DEFAULT_ERROR_STEP_ID = "tool_error"
DEFAULT_OK_OUTCOME = "ok"
DEFAULT_ERROR_OUTCOME = "error"
RUNTIME_ERROR_CAPABILITY = "wf.std.runtime_error"
__all__ = [
"DEFAULT_CALL_STEP_ID",
"DEFAULT_ERROR_OUTCOME",
"DEFAULT_ERROR_STEP_ID",
"DEFAULT_OK_OUTCOME",
"RUNTIME_ERROR_CAPABILITY",
]
src/wf_mcp/workflow_surface/handlers.py
+22 -9
View File
@@ -35,6 +35,13 @@ from wf_core import RuntimeContext
from ..events import make_event
from ..models import RawWorkflowPlan
from .constants import (
DEFAULT_CALL_STEP_ID,
DEFAULT_ERROR_OUTCOME,
DEFAULT_ERROR_STEP_ID,
DEFAULT_OK_OUTCOME,
RUNTIME_ERROR_CAPABILITY,
)
if TYPE_CHECKING:
from ..broker.service import WfMcpService
@@ -423,33 +430,39 @@ class WorkflowSurfaceHandlers:
title: str | None = None,
) -> dict[str, Any]:
"""Bootstrap the smallest patchable draft around one workflow capability."""
outcomes = self._outcomes_for_capability(capability_name) or ("ok",)
outcomes = self._outcomes_for_capability(capability_name) or (
DEFAULT_OK_OUTCOME,
)
steps: dict[str, Any] = {
"call": {
DEFAULT_CALL_STEP_ID: {
"use": capability_name,
"in": input_map,
"out": output_map,
}
}
routes: dict[str, dict[str, str]] = {"call": {"ok": "__end__"}}
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)
if "error" in outcomes and error_source is not None:
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
# exposes, a concrete state path that can become a runtime message.
steps["tool_error"] = {
"use": "wf.std.runtime_error",
steps[DEFAULT_ERROR_STEP_ID] = {
"use": RUNTIME_ERROR_CAPABILITY,
"in": {error_source: "message"},
"out": {},
}
routes["call"]["error"] = "tool_error"
routes["tool_error"] = {"ok": "__end__"}
routes[DEFAULT_CALL_STEP_ID][DEFAULT_ERROR_OUTCOME] = (
DEFAULT_ERROR_STEP_ID
)
routes[DEFAULT_ERROR_STEP_ID] = {DEFAULT_OK_OUTCOME: "__end__"}
draft = {
"name": name,
"input_schema": input_schema,
"state_schema": state_schema,
"output_schema": output_schema,
"start": "call",
"start": DEFAULT_CALL_STEP_ID,
"steps": steps,
"routes": routes,
}
src/wf_mcp/workflow_surface/models.py
+101 -1
View File
@@ -1,9 +1,46 @@
from __future__ import annotations
from typing import Any, Literal
from typing import Annotated, Any, Literal
from pydantic import BaseModel, Field
from wf_artifacts import ArtifactKind
from wf_artifacts.draft_workspaces.models import WORKSPACE_ID_PATTERN
WorkspaceId = Annotated[
str,
Field(
pattern=WORKSPACE_ID_PATTERN,
description="Draft workspace id. Use letters, numbers, underscore, dot, or dash.",
),
]
JsonSchemaObject = Annotated[
dict[str, Any],
Field(
description=(
"JSON Schema object. Keep this as ordinary JSON; nested schema fields "
"are passed through unchanged."
)
),
]
DraftPathMap = Annotated[
dict[str, str],
Field(
description=(
"Map local draft paths to workflow paths, for example "
"{'input.text': 'text'} or {'echoed': 'state.echoed'}."
)
),
]
JsonPatchOperations = Annotated[
list[dict[str, Any]],
Field(description="RFC 6902 JSON Patch operations."),
]
SourceBindings = Annotated[
dict[str, str],
Field(description="Map logical source ids to concrete source ids."),
]
class CallCapabilityResult(BaseModel):
"""Inspector-visible response contract for testing one workflow capability."""
@@ -52,3 +89,66 @@ class DraftWorkspaceResult(BaseModel):
default=None,
description="Full draft document, only returned when requested.",
)
class CreateDraftWorkspaceRequest(BaseModel):
"""Typed MCP request payload for creating a stored draft workspace."""
workspace_id: WorkspaceId
draft: dict[str, Any] = Field(description="WorkflowDraft JSON document.")
title: str | None = Field(default=None, description="Optional workspace title.")
class PatchDraftWorkspaceRequest(BaseModel):
"""Typed MCP request payload for revision-checked draft workspace patching."""
workspace_id: WorkspaceId
revision: int = Field(ge=1, description="Expected current workspace revision.")
patch: JsonPatchOperations
class CreateMinimalDraftWorkspaceRequest(BaseModel):
"""Typed MCP request payload for bootstrapping one-capability drafts."""
workspace_id: WorkspaceId
name: str = Field(description="Workflow draft name.")
capability_name: str = Field(
description="Workflow capability to call, such as demo.default.echo_tool."
)
input_schema: JsonSchemaObject
state_schema: JsonSchemaObject
output_schema: JsonSchemaObject
input_map: DraftPathMap
output_map: DraftPathMap
error_message_source: str | None = Field(
default=None,
description=(
"Optional state path used as runtime_error.message when the capability "
"has an error outcome, for example state.error_message."
),
)
title: str | None = Field(default=None, description="Optional workspace title.")
class CreateArtifactFromWorkspaceRequest(BaseModel):
"""Typed MCP request payload for saving a draft workspace as an artifact."""
workspace_id: WorkspaceId
artifact_id: str = Field(description="Immutable artifact id to write.")
version: int = Field(ge=1, description="Artifact version to write.")
title: str = Field(description="Human-readable artifact title.")
outcomes: list[str] = Field(description="Artifact-level outcomes.")
kind: ArtifactKind = Field(default="workflow", description="Artifact kind.")
description: str | None = Field(default=None, description="Optional description.")
required_capabilities: dict[str, dict[str, Any]] | None = Field(
default=None,
description="Optional explicit dependency contract override.",
)
source_bindings: SourceBindings | None = Field(
default=None,
description="Optional logical-to-concrete source bindings.",
)
created_from_catalog_version: str | None = Field(
default=None,
description="Optional catalog version used while authoring.",
)
src/wf_mcp/workflow_surface/tools.py
+40 -55
View File
@@ -9,7 +9,14 @@ from wf_artifacts.models import RequiredCapability
from wf_mcp.broker.service import WfMcpService
from .handlers import WorkflowSurfaceHandlers
from .models import CallCapabilityResult, DraftWorkspaceResult
from .models import (
CallCapabilityResult,
CreateArtifactFromWorkspaceRequest,
CreateDraftWorkspaceRequest,
CreateMinimalDraftWorkspaceRequest,
DraftWorkspaceResult,
PatchDraftWorkspaceRequest,
)
def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None:
@@ -196,15 +203,13 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
description="Store a mutable workflow draft workspace for iterative patching.",
)
async def create_draft_workspace(
workspace_id: str,
draft: dict[str, Any],
title: str | None = None,
request: CreateDraftWorkspaceRequest,
) -> DraftWorkspaceResult:
return DraftWorkspaceResult.model_validate(
await handlers.create_draft_workspace(
workspace_id=workspace_id,
draft=draft,
title=title,
workspace_id=request.workspace_id,
draft=request.draft,
title=request.title,
)
)
@@ -233,15 +238,13 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
),
)
async def patch_draft_workspace(
workspace_id: str,
revision: int,
patch: list[dict[str, Any]],
request: PatchDraftWorkspaceRequest,
) -> DraftWorkspaceResult:
return DraftWorkspaceResult.model_validate(
await handlers.patch_draft_workspace(
workspace_id=workspace_id,
revision=revision,
patch=patch,
workspace_id=request.workspace_id,
revision=request.revision,
patch=request.patch,
)
)
@@ -251,29 +254,20 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
description="Bootstrap a patchable draft workspace around one capability.",
)
async def create_minimal_draft_workspace(
workspace_id: str,
name: str,
capability_name: str,
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],
error_message_source: str | None = None,
title: str | None = None,
request: CreateMinimalDraftWorkspaceRequest,
) -> DraftWorkspaceResult:
return DraftWorkspaceResult.model_validate(
await handlers.create_minimal_draft_workspace(
workspace_id=workspace_id,
name=name,
capability_name=capability_name,
input_schema=input_schema,
state_schema=state_schema,
output_schema=output_schema,
input_map=input_map,
output_map=output_map,
error_message_source=error_message_source,
title=title,
workspace_id=request.workspace_id,
name=request.name,
capability_name=request.capability_name,
input_schema=request.input_schema,
state_schema=request.state_schema,
output_schema=request.output_schema,
input_map=request.input_map,
output_map=request.output_map,
error_message_source=request.error_message_source,
title=request.title,
)
)
@@ -286,38 +280,29 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
),
)
async def create_artifact_from_workspace(
workspace_id: str,
artifact_id: str,
version: int,
title: str,
outcomes: list[str],
kind: ArtifactKind = "workflow",
description: str | None = None,
required_capabilities: (
Mapping[str, RequiredCapability | dict[str, Any]] | None
) = None,
source_bindings: Mapping[str, str] | None = None,
created_from_catalog_version: str | None = None,
request: CreateArtifactFromWorkspaceRequest,
) -> dict[str, Any]:
return await handlers.create_artifact_from_workspace(
workspace_id=workspace_id,
artifact_id=artifact_id,
version=version,
title=title,
kind=kind,
description=description,
outcomes=outcomes,
workspace_id=request.workspace_id,
artifact_id=request.artifact_id,
version=request.version,
title=request.title,
kind=request.kind,
description=request.description,
outcomes=request.outcomes,
required_capabilities={
name: (
capability.model_dump()
if isinstance(capability, RequiredCapability)
else capability
)
for name, capability in (required_capabilities or {}).items()
for name, capability in (
request.required_capabilities or {}
).items()
}
or None,
source_bindings=dict(source_bindings or {}),
created_from_catalog_version=created_from_catalog_version,
source_bindings=dict(request.source_bindings or {}),
created_from_catalog_version=request.created_from_catalog_version,
)
@server.tool(