some more ops

could we make subgraphs as ops or patterns like this?

src/wf_authoring/__init__.py

@@ -22,10 +22,15 @@ from .dsl import (
 from .ops import (
     BoolOutput,
     CoalesceInput,
+    ConcatInput,
     ConstantInput,
     CountOutput,
+    ExtractFieldInput,
+    FilterItemsInput,
+    FilterItemsPresentInput,
     ItemOutput,
     MappingOutput,
+    MappingItemsOutput,
     MaybeItemOutput,
     PickPathInput,
     PickKeyInput,
@@ -33,11 +38,16 @@ from .ops import (
     RenameFieldsInput,
     RuntimeErrorInput,
     SequenceInput,
+    TextOutput,
     TruthyInput,
     ValueOutput,
     coalesce,
+    concat,
     constant,
     default_if_none,
+    extract_field,
+    filter_items,
+    filter_items_present,
     first_item,
     first_item_maybe,
     first_item_or_none,
@@ -75,11 +85,16 @@ __all__ = [
     "async_subgraph_node",
     "BoolOutput",
     "CoalesceInput",
+    "ConcatInput",
     "ConstantInput",
     "CountOutput",
+    "ExtractFieldInput",
+    "FilterItemsInput",
+    "FilterItemsPresentInput",
     "GraphPath",
     "ItemOutput",
     "MappingOutput",
+    "MappingItemsOutput",
     "MaybeItemOutput",
     "PickKeyInput",
     "PickPathInput",
@@ -95,6 +110,7 @@ __all__ = [
     "SyncRegistryHandler",
     "SequenceInput",
     "StateFieldMetadata",
+    "TextOutput",
     "TruthyInput",
     "ValueOutput",
     "WorkflowBuilder",
@@ -103,8 +119,12 @@ __all__ = [
     "build_registry",
     "bind_state",
     "coalesce",
+    "concat",
     "constant",
     "default_if_none",
+    "extract_field",
+    "filter_items",
+    "filter_items_present",
     "merge_maps",
     "context",
     "context_path",

src/wf_authoring/ops/__init__.py

@@ -1,9 +1,17 @@
 from .sequences import (
     BoolOutput,
     CountOutput,
+    ExtractFieldInput,
+    FilterItemsInput,
+    FilterItemsPresentInput,
     ItemOutput,
+    MappingItemsOutput,
     MaybeItemOutput,
     SequenceInput,
+    ValuesOutput,
+    extract_field,
+    filter_items,
+    filter_items_present,
     first_item,
     first_item_maybe,
     first_item_or_none,
@@ -14,6 +22,7 @@ from .sequences import (
 )
 from .values import (
     CoalesceInput,
+    ConcatInput,
     ConstantInput,
     MappingOutput,
     PickPathInput,
@@ -22,8 +31,10 @@ from .values import (
     RenameFieldsInput,
     RuntimeErrorInput,
     TruthyInput,
+    TextOutput,
     ValueOutput,
     coalesce,
+    concat,
     constant,
     default_if_none,
     pick_path,
@@ -37,10 +48,15 @@ from .values import (
 __all__ = [
     "BoolOutput",
     "CoalesceInput",
+    "ConcatInput",
     "ConstantInput",
     "CountOutput",
+    "ExtractFieldInput",
+    "FilterItemsInput",
+    "FilterItemsPresentInput",
     "ItemOutput",
     "MappingOutput",
+    "MappingItemsOutput",
     "MaybeItemOutput",
     "PickPathInput",
     "PickKeyInput",
@@ -49,10 +65,16 @@ __all__ = [
     "RuntimeErrorInput",
     "SequenceInput",
     "TruthyInput",
+    "TextOutput",
     "ValueOutput",
+    "ValuesOutput",
     "coalesce",
+    "concat",
     "constant",
     "default_if_none",
+    "extract_field",
+    "filter_items",
+    "filter_items_present",
     "first_item",
     "first_item_maybe",
     "first_item_or_none",

src/wf_authoring/ops/sequences.py

@@ -37,6 +37,40 @@ class BoolOutput(BaseModel):
     value: bool
 
 
+class FilterItemsInput(BaseModel):
+    """Input model for filtering mapping items by exact key/value match."""
+
+    items: list[dict[str, Any]]
+    key: str
+    value: Any
+
+
+class FilterItemsPresentInput(BaseModel):
+    """Input model for filtering mapping items that contain a key."""
+
+    items: list[dict[str, Any]]
+    key: str
+
+
+class MappingItemsOutput(BaseModel):
+    """Output model for ops that return mapping items."""
+
+    items: list[dict[str, Any]]
+
+
+class ExtractFieldInput(BaseModel):
+    """Input model for extracting one field from mapping items."""
+
+    items: list[dict[str, Any]]
+    field: str
+
+
+class ValuesOutput(BaseModel):
+    """Output model for ops that return arbitrary values."""
+
+    values: list[Any]
+
+
 @node(
     name="authoring.first_item",
     input_model=SequenceInput,
@@ -119,3 +153,40 @@ def length(input: SequenceInput) -> CountOutput:
 def is_empty(input: SequenceInput) -> BoolOutput:
     """Return whether a sequence has no items."""
     return BoolOutput(value=not input.items)
+
+
+@node(
+    name="authoring.filter_items",
+    input_model=FilterItemsInput,
+    output_model=MappingItemsOutput,
+    description="Filter mapping items by exact key/value match.",
+)
+def filter_items(input: FilterItemsInput) -> MappingItemsOutput:
+    """Return items where item[key] exactly equals value."""
+    return MappingItemsOutput(
+        items=[item for item in input.items if item.get(input.key) == input.value]
+    )
+
+
+@node(
+    name="authoring.filter_items_present",
+    input_model=FilterItemsPresentInput,
+    output_model=MappingItemsOutput,
+    description="Filter mapping items to those containing the requested key.",
+)
+def filter_items_present(input: FilterItemsPresentInput) -> MappingItemsOutput:
+    """Return items that contain key, regardless of the stored value."""
+    return MappingItemsOutput(items=[item for item in input.items if input.key in item])
+
+
+@node(
+    name="authoring.extract_field",
+    input_model=ExtractFieldInput,
+    output_model=ValuesOutput,
+    description="Extract one field from each mapping item that contains it.",
+)
+def extract_field(input: ExtractFieldInput) -> ValuesOutput:
+    """Return item[field] for each item containing field."""
+    return ValuesOutput(
+        values=[item[input.field] for item in input.items if input.field in item]
+    )

src/wf_authoring/ops/values.py

@@ -73,6 +73,19 @@ class RuntimeErrorInput(BaseModel):
     details: dict[str, Any] = Field(default_factory=dict)
 
 
+class ConcatInput(BaseModel):
+    """Input model for joining string values."""
+
+    items: list[str]
+    separator: str = ""
+
+
+class TextOutput(BaseModel):
+    """Output model for ops that emit text."""
+
+    text: str
+
+
 @node(
     name="authoring.coalesce",
     input_model=CoalesceInput,
@@ -180,6 +193,17 @@ def truthy(input: TruthyInput) -> NodeReturn[ValueOutput]:
     return NodeReturn(outcome=outcome, output=ValueOutput(value=value))
 
 
+@node(
+    name="authoring.concat",
+    input_model=ConcatInput,
+    output_model=TextOutput,
+    description="Join string items using separator.",
+)
+def concat(input: ConcatInput) -> TextOutput:
+    """Join string items using separator."""
+    return TextOutput(text=input.separator.join(input.items))
+
+
 @node(
     name="authoring.runtime_error",
     input_model=RuntimeErrorInput,

src/wf_mcp/broker/service/builtins.py

@@ -2,7 +2,8 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING, Any
 
-from wf_authoring import NodeSpec, coalesce, constant, default_if_none, first_item
+from wf_authoring import NodeSpec, coalesce, concat, constant, default_if_none
+from wf_authoring import extract_field, filter_items, filter_items_present, first_item
 from wf_authoring import first_item_maybe, first_item_or_none, is_empty, last_item
 from wf_authoring import last_item_or_none, length, node, pick_key, pick_path
 from wf_authoring import project_fields, rename_fields, runtime_error, truthy
@@ -43,6 +44,10 @@ AUTHORING_STD_SPECS: tuple[NodeSpec[Any, Any], ...] = (
     last_item_or_none,
     length,
     is_empty,
+    filter_items,
+    filter_items_present,
+    extract_field,
+    concat,
 )
 """Existing authoring ops that are also exposed through the workflow stdlib."""