look at that users of @node. first, three variants

This commit is contained in:
lda
2026-05-06 22:00:17 +07:00 Verified
parent c4a68135fd
commit 93fe9da816
8 changed files with 287 additions and 25 deletions
+18 -2
View File
@@ -5,7 +5,7 @@
Main packages live under `src/`:
- `wf_core`: workflow model, validation, runtime semantics, frames, trace, interrupts, foreach, async execution.
- `wf_authoring`: ergonomic authoring layer including `@node`, `NodeSpec`, `WorkflowBuilder`, conditions, paths, and subgraph wrapping.
- `wf_mcp`: MCP broker/proxy layer for managing multiple backend MCP connections, discovery/catalog snapshots, transparent FastMCP proxying, config/admin tools, and eventual workflow build/run integration.
- `wf_mcp`: MCP broker/proxy layer for managing multiple backend MCP connections, transparent FastMCP proxying, live config/admin tools, hot reload, tool introspection, discovery/catalog snapshots, and eventual workflow build/run integration.
Important docs:
- `readme.md`: running design notes and architecture.
@@ -13,4 +13,20 @@ Important docs:
- `wf_mcp_plan.md`: MCP proxy/broker/workflow integration plan.
- `scratchpad.md`: rough design history.
Current MCP direction: transparent proxy mode is the main product path. Old broker tools remain useful for debugging/admin/catalog operations, but protocol-native FastMCP proxying exposes upstream tools/resources/prompts as first-class MCP capabilities.
Current MCP direction:
- Transparent proxy mode is the main product path.
- Old broker mode remains useful for debugging/admin/catalog operations.
- Protocol-native FastMCP proxying exposes upstream tools/resources/prompts as first-class MCP capabilities.
- Current configured live connections have included `context7.default`, `serena.default`, and `everything.default`; treat `wf_mcp.config.json` as user-owned live state.
- Direct Serena is configured outside `wf_mcp` and should be preferred for code navigation/editing because it does not reset when `wf-mcp` hot-reloads.
Recent `wf_mcp` capabilities:
- Pydantic config boundary in `config_models.py`.
- Config mutation boundary in `config_manager.py`.
- Proxy validation in `proxy_validation.py`.
- Transparent proxy runtime and manual hot reload in `transparent_proxy.py`.
- Explicit name mapping in `names.py`.
- Opaque cursor pagination helpers in `pagination.py`.
- Admin tools under `wf.mcp_*`: list/get config, add/update/enable/disable/remove connection, reload config, list/get proxy tools.
- `wf.mcp_list_proxy_tools` supports `connection_id`, `query`, `limit`, and `cursor`; returns `{tools, nextCursor, total}`.
- `wf.mcp_get_proxy_tool` returns one detailed proxied tool row with schema where available.
+12 -2
View File
@@ -7,14 +7,24 @@ General:
- Keep MCP proxy concepts separate from workflow-specific concepts like `outcome`.
- Do not leak workflow-only fields into MCP `tools/list`.
MCP/proxy conventions:
- Transparent proxy mode is the product path; old broker mode is secondary/debug/admin-oriented.
- Admin tools live under the reserved namespace `wf.mcp_*`.
- Upstream FastMCP names use Namespace behavior: `<connection_id>_<local_tool_name>`, e.g. `everything.default_echo`.
- Keep name parsing and unmangling in `names.py`; do not scatter string slicing across modules.
- Keep cursor mechanics in `pagination.py`; use opaque cursors and `nextCursor` to mirror MCP pagination style.
- Keep config disk writes in `config_manager.py`; transparent proxy code should expose/administer behavior, not own JSON mutation details.
- `wf_mcp.config.json` is user-owned live config. Do not edit/revert it unless explicitly asked.
Code style:
- Use precise type hints and modern Python collection syntax (`list[str]`, `dict[str, Any]`).
- Keep modules layered by responsibility; avoid stuffing everything into service/runtime files.
- Prefer small helper modules when behavior becomes a boundary (`config_models.py`, `config_manager.py`, `proxy_validation.py`).
- Prefer small helper modules when behavior becomes a boundary (`config_models.py`, `config_manager.py`, `proxy_validation.py`, `names.py`, `pagination.py`).
- Validation should fail early with clear errors.
- Tests should exercise behavior through public APIs/MCP calls where practical.
- For MCP client `CallToolResult.structured_content`, guard for `None` in tests before subscripting; use helper assertions where useful.
Editing rules from repo collaboration:
- Use `apply_patch` for manual code edits.
- Do not revert user-owned changes.
- Treat `wf_mcp.config.json` as user-owned live config unless explicitly asked.
- Direct Serena MCP is available and can be used for semantic navigation/editing; onboarding is already complete.
+15 -2
View File
@@ -4,11 +4,12 @@ Use PowerShell on Windows from the repo root.
Testing:
- `uv run --with pytest pytest -q`
- Focused example: `uv run --with pytest pytest tests/test_wf_mcp_transparent_proxy.py -q`
- Focused MCP proxy tests: `uv run --with pytest pytest tests/test_wf_mcp_transparent_proxy.py -q`
- Focused names/pagination examples: `uv run --with pytest pytest tests/test_wf_mcp_names.py tests/test_wf_mcp_transparent_proxy.py -q`
Lint/type checks:
- `uv run ruff check src/wf_mcp tests`
- Focused basedpyright example: `uv run basedpyright src/wf_mcp/transparent_proxy.py --level error`
- Focused basedpyright example: `uv run basedpyright src/wf_mcp/transparent_proxy.py src/wf_mcp/pagination.py --level error`
Formatting:
- `uv run ruff format`
@@ -19,6 +20,18 @@ CLI / MCP server:
- Old broker mode: `uv run wf-mcp --config wf_mcp.config.json serve --mode broker`
- Optional compatibility/search flags: `--resources-as-tools`, `--prompts-as-tools`, `--search-tools`
Useful live MCP admin tools exposed by `wf-mcp`:
- `wf.mcp_list_connections`
- `wf.mcp_get_config`
- `wf.mcp_add_connection`
- `wf.mcp_update_connection`
- `wf.mcp_enable_connection`
- `wf.mcp_disable_connection`
- `wf.mcp_remove_connection`
- `wf.mcp_reload_config`
- `wf.mcp_list_proxy_tools`
- `wf.mcp_get_proxy_tool`
Useful Windows shell commands:
- Fast search: `rg "pattern" path`
- List files: `Get-ChildItem -Force`
@@ -6,7 +6,12 @@ Before considering a code task done:
- Run ruff on touched source/tests, usually `uv run ruff check src/wf_mcp tests` for MCP work.
- Run focused basedpyright at error level for new or heavily changed files when type issues are likely.
- Check `git status --short` and distinguish user-owned config changes from code changes.
- For MCP/proxy changes, consider direct FastMCP client verification when Codex's native MCP tool registry is stale.
- Summarize functional changes and verification results concisely.
Known environment note:
- Windows sandbox may block commands with `CreateProcessAsUserW failed: 5`; retry important commands with escalation rather than working around via unsafe shell tricks.
Recent known-good full-suite count after proxy tool pagination/detail work:
- `48 passed, 1 skipped`
Known environment notes:
- Windows sandbox may block commands with `CreateProcessAsUserW failed: 5`; retry important commands with escalation rather than working around via unsafe shell tricks.
- Codex/native MCP tool schemas may not refresh dynamically after `wf-mcp` hot reload. A fresh client/session may be needed to see newly added MCP tools.
+14
View File
@@ -2,6 +2,14 @@ from .builder import WorkflowBuilder
from .catalog import NodeCatalog, NodeCatalogEntry
from .conditions import context, exists, expr, input, state
from .mapping import bind_fields, bind_state, merge_maps
from .ops import (
ItemOutput,
MaybeItemOutput,
SequenceInput,
first_item,
first_item_maybe,
first_item_or_none,
)
from .paths import GraphPath, context_path, graph_path, input_path, state_path
from .spec import (
AsyncRegistryHandler,
@@ -18,10 +26,13 @@ __all__ = [
"NodeCatalog",
"NodeCatalogEntry",
"GraphPath",
"ItemOutput",
"MaybeItemOutput",
"NodeReturn",
"NodeSpec",
"AsyncRegistryHandler",
"SyncRegistryHandler",
"SequenceInput",
"WorkflowBuilder",
"bind_fields",
"build_async_registry",
@@ -32,6 +43,9 @@ __all__ = [
"context_path",
"expr",
"exists",
"first_item",
"first_item_maybe",
"first_item_or_none",
"graph_path",
"input",
"input_path",
+56
View File
@@ -0,0 +1,56 @@
from __future__ import annotations
from typing import Any
from pydantic import BaseModel
from .spec import NodeReturn, node
class SequenceInput(BaseModel):
items: list[Any]
class ItemOutput(BaseModel):
item: Any
class MaybeItemOutput(BaseModel):
item: Any | None = None
@node(
name="authoring.first_item",
input_model=SequenceInput,
output_model=ItemOutput,
description="Select the first item from a non-empty sequence.",
)
def first_item(input: SequenceInput) -> ItemOutput:
if not input.items:
raise ValueError("first_item requires at least one item")
return ItemOutput(item=input.items[0])
@node(
name="authoring.first_item_or_none",
input_model=SequenceInput,
output_model=ItemOutput,
description="Select the first item from a sequence, or None when it is empty.",
)
def first_item_or_none(input: SequenceInput) -> ItemOutput:
return ItemOutput(item=input.items[0] if input.items else None)
@node(
name="authoring.first_item_maybe",
input_model=SequenceInput,
output_model=MaybeItemOutput,
outcomes=("found", "missing"),
description=(
"Select the first item from a sequence, routing to found or missing."
),
)
def first_item_maybe(input: SequenceInput) -> NodeReturn[MaybeItemOutput]:
if not input.items:
return NodeReturn(outcome="missing", output=MaybeItemOutput())
return NodeReturn(outcome="found", output=MaybeItemOutput(item=input.items[0]))
+49 -17
View File
@@ -21,10 +21,30 @@ from wf_core import NodeDef, RuntimeContext, SchemaRef
InputT = TypeVar("InputT", bound=BaseModel)
OutputT = TypeVar("OutputT", bound=BaseModel)
NodeCallable = Callable[[InputT, RuntimeContext], "NodeReturn[OutputT] | OutputT"]
AsyncNodeCallable = Callable[
ContextNodeCallable = Callable[
[InputT, RuntimeContext], "NodeReturn[OutputT] | OutputT"
]
PlainNodeCallable = Callable[[InputT], "NodeReturn[OutputT] | OutputT"]
NodeCallable = (
Callable[[InputT, RuntimeContext], "NodeReturn[OutputT] | OutputT"]
| Callable[[InputT], "NodeReturn[OutputT] | OutputT"]
)
AsyncContextNodeCallable = Callable[
[InputT, RuntimeContext], Awaitable["NodeReturn[OutputT] | OutputT"]
]
AsyncPlainNodeCallable = Callable[
[InputT], Awaitable["NodeReturn[OutputT] | OutputT"]
]
AsyncNodeCallable = (
Callable[
[InputT, RuntimeContext],
Awaitable["NodeReturn[OutputT] | OutputT"],
]
| Callable[[InputT], Awaitable["NodeReturn[OutputT] | OutputT"]]
)
SyncRegistryHandler = Callable[[dict[str, Any], RuntimeContext], dict[str, Any]]
AsyncRegistryHandler = Callable[
[dict[str, Any], RuntimeContext], Awaitable[dict[str, Any]]
@@ -76,30 +96,31 @@ def _infer_models(
) -> tuple[type[BaseModel], type[BaseModel]]:
hints = get_type_hints(fn, include_extras=True)
params = list(signature(fn).parameters.values())
if len(params) < 2:
raise TypeError("node function must accept at least (payload, ctx) parameters")
if len(params) not in {1, 2}:
raise TypeError("node function must accept (payload) or (payload, ctx)")
payload_param = params[0]
ctx_param = params[1]
if payload_param.kind not in (
Parameter.POSITIONAL_ONLY,
Parameter.POSITIONAL_OR_KEYWORD,
):
raise TypeError("node payload parameter must be positional")
if ctx_param.kind not in (
Parameter.POSITIONAL_ONLY,
Parameter.POSITIONAL_OR_KEYWORD,
):
raise TypeError("node context parameter must be positional")
input_model = hints.get(payload_param.name)
if not _is_basemodel_subclass(input_model):
raise TypeError("node payload annotation must be a pydantic BaseModel subclass")
ctx_type = hints.get(ctx_param.name)
if ctx_type is not RuntimeContext:
raise TypeError("node context annotation must be wf_core.RuntimeContext")
if len(params) == 2:
ctx_param = params[1]
if ctx_param.kind not in (
Parameter.POSITIONAL_ONLY,
Parameter.POSITIONAL_OR_KEYWORD,
):
raise TypeError("node context parameter must be positional")
ctx_type = hints.get(ctx_param.name)
if ctx_type is not RuntimeContext:
raise TypeError("node context annotation must be wf_core.RuntimeContext")
return_type = hints.get("return")
if return_type is None:
@@ -122,6 +143,10 @@ def _infer_models(
)
def _accepts_context(fn: Callable[..., object]) -> bool:
return len(signature(fn).parameters) >= 2
@dataclass(slots=True)
class NodeSpec(Generic[InputT, OutputT]):
name: str
@@ -131,13 +156,18 @@ class NodeSpec(Generic[InputT, OutputT]):
fn: NodeCallable[InputT, OutputT] | AsyncNodeCallable[InputT, OutputT]
description: str | None = None
is_async: bool = False
accepts_context: bool = True
def __call__(
self,
payload: InputT,
ctx: RuntimeContext,
ctx: RuntimeContext | None = None,
) -> NodeReturn[OutputT] | OutputT | Awaitable[NodeReturn[OutputT] | OutputT]:
return self.fn(payload, ctx)
if self.accepts_context:
if ctx is None:
raise TypeError(f"node {self.name!r} requires RuntimeContext")
return cast("ContextNodeCallable[InputT, OutputT]", self.fn)(payload, ctx)
return cast("PlainNodeCallable[InputT, OutputT]", self.fn)(payload)
def to_node_def(self) -> NodeDef:
return NodeDef(
@@ -155,7 +185,7 @@ class NodeSpec(Generic[InputT, OutputT]):
def handler(payload: dict[str, Any], ctx: RuntimeContext) -> dict[str, Any]:
parsed = self.input_model.model_validate(payload)
raw = self.fn(parsed, ctx)
raw = self(parsed, ctx)
return _coerce_registry_result(
node_name=self.name,
output_model=self.output_model,
@@ -171,7 +201,7 @@ class NodeSpec(Generic[InputT, OutputT]):
ctx: RuntimeContext,
) -> dict[str, Any]:
parsed = self.input_model.model_validate(payload)
raw_result = self.fn(parsed, ctx)
raw_result = self(parsed, ctx)
if self.is_async:
raw = await cast(
Awaitable[NodeReturn[OutputT] | OutputT],
@@ -245,6 +275,7 @@ def node(
resolved_name = name or getattr(fn, "__name__", "node")
resolved_is_async = iscoroutinefunction(fn) if is_async is None else is_async
resolved_accepts_context = _accepts_context(fn)
return cast(
NodeSpec[InputT, OutputT],
NodeSpec(
@@ -255,6 +286,7 @@ def node(
fn=cast(Any, fn),
description=description or fn.__doc__,
is_async=resolved_is_async,
accepts_context=resolved_accepts_context,
),
)
+116
View File
@@ -0,0 +1,116 @@
from __future__ import annotations
import pytest
from wf_authoring import (
WorkflowBuilder,
bind_fields,
bind_state,
build_registry,
first_item,
first_item_maybe,
first_item_or_none,
state_path,
)
from wf_core import RunStatus, SchemaRef, StateField, StateSchema, execute_workflow
def _build_first_workflow(use_safe_first: bool = False):
spec = first_item_or_none if use_safe_first else first_item
builder = WorkflowBuilder(
name="first_demo",
input_schema=SchemaRef(type="object"),
state_schema=StateSchema(
fields={
"items": StateField(type="array"),
"item": StateField(type="object"),
}
),
output_schema=SchemaRef(type="object"),
start="pick_first",
)
node = builder.use(
spec,
id="pick_first",
in_map=bind_fields(items=state_path("items")),
out_map=bind_state(item=state_path("item")),
)
builder.connect(node, "ok", "__end__")
return builder.compile(), build_registry(spec)
def _build_first_maybe_workflow():
builder = WorkflowBuilder(
name="first_maybe_demo",
input_schema=SchemaRef(type="object"),
state_schema=StateSchema(
fields={
"items": StateField(type="array"),
"item": StateField(type="object"),
"missing": StateField(type="boolean"),
}
),
output_schema=SchemaRef(type="object"),
start="pick_first",
)
pick_first = builder.use(
first_item_maybe,
id="pick_first",
in_map=bind_fields(items=state_path("items")),
out_map=bind_state(item=state_path("item")),
)
mark_missing = builder.use(
first_item_or_none,
id="mark_missing",
in_map=bind_fields(items=state_path("items")),
out_map=bind_state(item=state_path("item")),
)
builder.connect(pick_first, "found", "__end__")
builder.connect(pick_first, "missing", mark_missing)
builder.connect(mark_missing, "ok", "__end__")
return builder.compile(), build_registry(first_item_maybe, first_item_or_none)
def test_first_item_selects_first_value_through_workflow() -> None:
workflow, registry = _build_first_workflow()
run = execute_workflow(workflow, {"items": ["a", "b"]}, registry)
assert run.status == RunStatus.COMPLETED
assert run.state["item"] == "a"
def test_first_item_fails_on_empty_sequence() -> None:
workflow, registry = _build_first_workflow()
with pytest.raises(ValueError, match="first_item requires at least one item"):
execute_workflow(workflow, {"items": []}, registry)
def test_first_item_or_none_returns_none_for_empty_sequence() -> None:
workflow, registry = _build_first_workflow(use_safe_first=True)
run = execute_workflow(workflow, {"items": []}, registry)
assert run.status == RunStatus.COMPLETED
assert run.state["item"] is None
def test_first_item_maybe_routes_found_outcome() -> None:
workflow, registry = _build_first_maybe_workflow()
run = execute_workflow(workflow, {"items": ["a", "b"]}, registry)
assert run.status == RunStatus.COMPLETED
assert run.state["item"] == "a"
assert run.trace[0].outcome == "found"
def test_first_item_maybe_routes_missing_outcome() -> None:
workflow, registry = _build_first_maybe_workflow()
run = execute_workflow(workflow, {"items": []}, registry)
assert run.status == RunStatus.COMPLETED
assert run.state["item"] is None
assert run.trace[0].outcome == "missing"