first recipe: extract textcontent from output

src/wf_authoring/__init__.py

@@ -75,6 +75,14 @@ from .nodes import (
     outcome,
 )
 from .reducers import AuthoredReducer, ReducerCatalog, reducer
+from .recipes import (
+    ExtractTextContentInput,
+    ExtractTextContentOutput,
+    ExtractTextContentState,
+    build_extract_text_content_spec,
+    build_extract_text_content_workflow,
+    extract_text_content,
+)
 from .schemas import StateFieldMetadata, state_field
 from .subgraph import async_subgraph_node, subgraph_node, subgraph_ref
 
@@ -89,6 +97,9 @@ __all__ = [
     "ConstantInput",
     "CountOutput",
     "ExtractFieldInput",
+    "ExtractTextContentInput",
+    "ExtractTextContentOutput",
+    "ExtractTextContentState",
     "FilterItemsInput",
     "FilterItemsPresentInput",
     "GraphPath",
@@ -117,12 +128,15 @@ __all__ = [
     "bind_fields",
     "build_async_registry",
     "build_registry",
+    "build_extract_text_content_spec",
+    "build_extract_text_content_workflow",
     "bind_state",
     "coalesce",
     "concat",
     "constant",
     "default_if_none",
     "extract_field",
+    "extract_text_content",
     "filter_items",
     "filter_items_present",
     "merge_maps",

src/wf_authoring/recipes.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel
+
+from wf_authoring.builder import WorkflowBuilder
+from wf_authoring.dsl import input_from, input_path, input_value, output_to, state_path
+from wf_authoring.nodes import NodeSpec, build_registry
+from wf_authoring.ops import concat, extract_field, filter_items
+from wf_authoring.subgraph import subgraph_node
+from wf_core import END
+
+
+class ExtractTextContentInput(BaseModel):
+    """Input for extracting text from generic content-block dictionaries."""
+
+    content: list[dict[str, Any]]
+    separator: str = ""
+
+
+class ExtractTextContentState(BaseModel):
+    """Internal recipe state for staged content extraction."""
+
+    text_items: list[dict[str, Any]] = []
+    texts: list[str] = []
+    text: str = ""
+
+
+class ExtractTextContentOutput(BaseModel):
+    """Output text joined from all text content blocks."""
+
+    text: str
+
+
+def build_extract_text_content_workflow():
+    """Build the first-party text-content extraction recipe workflow.
+
+    Current limitation: this recipe is exposed as a wrapper-node capability via
+    `subgraph_node`, not as a native `SubgraphNode`. Callers can use it like any
+    other NodeSpec, but parent traces see one node call and child frames,
+    interrupts, and step-level diagnostics are not promoted to the parent run.
+    """
+    builder = WorkflowBuilder(
+        name="extract_text_content",
+        input_schema=ExtractTextContentInput,
+        state_schema=ExtractTextContentState,
+        output_schema=ExtractTextContentOutput,
+    )
+    filter_text = builder.use(
+        filter_items,
+        id="filter_text",
+        input=[
+            input_from(input_path("content"), "items"),
+            input_value("key", "type"),
+            input_value("value", "text"),
+        ],
+        output=[output_to("items", state_path("text_items"))],
+    )
+    extract_text = builder.use(
+        extract_field,
+        id="extract_text",
+        input=[
+            input_from(state_path("text_items"), "items"),
+            input_value("field", "text"),
+        ],
+        output=[output_to("values", state_path("texts"))],
+    )
+    join_text = builder.use(
+        concat,
+        id="join_text",
+        input=[
+            input_from(state_path("texts"), "items"),
+            input_from(input_path("separator"), "separator"),
+        ],
+        output=[output_to("text", state_path("text"))],
+    )
+    builder.set_entry_point(filter_text)
+    builder.connect(filter_text, "ok", extract_text)
+    builder.connect(extract_text, "ok", join_text)
+    builder.connect(join_text, "ok", END)
+    return builder.compile()
+
+
+def build_extract_text_content_spec() -> NodeSpec[
+    ExtractTextContentInput, ExtractTextContentOutput
+]:
+    """Return the recipe as a normal NodeSpec for first-party capability sources."""
+    workflow = build_extract_text_content_workflow()
+    return subgraph_node(
+        name="authoring.extract_text_content",
+        workflow=workflow,
+        registry=build_registry(filter_items, extract_field, concat),
+        input_model=ExtractTextContentInput,
+        output_model=ExtractTextContentOutput,
+        description=(
+            "Extract text fields from content blocks with type='text' and join "
+            "them using separator."
+        ),
+    )
+
+
+extract_text_content = build_extract_text_content_spec()
+"""First-party recipe composed from generic sequence/value ops."""

src/wf_mcp/broker/service/builtins.py

@@ -7,6 +7,7 @@ from wf_authoring import extract_field, filter_items, filter_items_present, firs
 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
+from wf_authoring import extract_text_content
 from wf_core.runtime.ops.merges import DEFAULT_REDUCER_DEFINITIONS
 
 from wf_platform import (
@@ -26,6 +27,9 @@ BUILTIN_CONNECTION_ID = "wf.std"
 MCP_SOURCE_ID = "wf.mcp"
 """Reserved source id for future workflow-safe MCP utility node specs."""
 
+RECIPE_SOURCE_ID = "wf.recipes"
+"""Internal source id for first-party composed workflow recipes."""
+
 
 AUTHORING_STD_SPECS: tuple[NodeSpec[Any, Any], ...] = (
     coalesce,
@@ -52,16 +56,36 @@ AUTHORING_STD_SPECS: tuple[NodeSpec[Any, Any], ...] = (
 """Existing authoring ops that are also exposed through the workflow stdlib."""
 
 
-def builtin_specs() -> dict[str, NodeSpec[Any, Any]]:
+RECIPE_SPECS: tuple[NodeSpec[Any, Any], ...] = (extract_text_content,)
-    """Return built-in NodeSpecs available to raw broker workflow plans."""
+"""Composed first-party recipes exposed as capabilities."""
-    specs = [
+
-        node(spec, name=spec.name.removeprefix("authoring."))
+
-        for spec in AUTHORING_STD_SPECS
+def _qualified_specs(
+    source_id: str,
+    specs: tuple[NodeSpec[Any, Any], ...],
+) -> dict[str, NodeSpec[Any, Any]]:
+    """Return specs with authoring names rewritten under one source id."""
+    local_specs = [
+        node(spec, name=spec.name.removeprefix("authoring.")) for spec in specs
     ]
-    qualified_specs = [qualify_spec(BUILTIN_CONNECTION_ID, spec) for spec in specs]
+    qualified_specs = [qualify_spec(source_id, spec) for spec in local_specs]
     return {spec.name: spec for spec in qualified_specs}
 
 
+def builtin_specs() -> dict[str, NodeSpec[Any, Any]]:
+    """Return primitive built-in NodeSpecs available to raw broker workflow plans."""
+    return _qualified_specs(BUILTIN_CONNECTION_ID, AUTHORING_STD_SPECS)
+
+
+def recipe_specs() -> dict[str, NodeSpec[Any, Any]]:
+    """Return composed first-party recipe specs.
+
+    Recipes are wrapper-node subgraphs today. They are useful workflow-facing
+    capabilities, but parent runs do not yet see their child graph frames.
+    """
+    return _qualified_specs(RECIPE_SOURCE_ID, RECIPE_SPECS)
+
+
 def builtin_reducers() -> dict[str, ReducerSpec]:
     """Return built-in reducers owned by the workflow standard library."""
     return {
@@ -94,4 +118,16 @@ def builtin_sources() -> dict[str, CapabilitySource]:
             permissions=SourcePermissions(safe_for_workflow=True),
             description="Workflow standard-library nodes.",
         ),
+        RECIPE_SOURCE_ID: CapabilitySource(
+            id=RECIPE_SOURCE_ID,
+            kind="system",
+            capabilities=CapabilityBuckets(node_specs=recipe_specs()),
+            visibility=SourceVisibility(
+                planner=True,
+                mcp_client=True,
+                admin_dashboard=True,
+            ),
+            permissions=SourcePermissions(safe_for_workflow=True),
+            description="First-party workflow recipes composed from standard nodes.",
+        ),
     }

tests/authoring/test_ops.py

@@ -10,6 +10,7 @@ from wf_authoring import (
     default_if_none,
     concat,
     extract_field,
+    extract_text_content,
     filter_items,
     filter_items_present,
     first_item,
@@ -381,6 +382,25 @@ def test_concat_joins_strings_with_separator() -> None:
     assert result["output"]["text"] == "a\nb\nc"
 
 
+def test_extract_text_content_recipe_filters_extracts_and_joins_text_blocks() -> None:
+    registry = build_registry(extract_text_content)
+
+    result = registry["authoring.extract_text_content"](
+        {
+            "content": [
+                {"type": "text", "text": "hello"},
+                {"type": "image", "url": "img://1"},
+                {"type": "text", "text": "world"},
+            ],
+            "separator": " ",
+        },
+        RuntimeContext(current_node_id="extract_text_content"),
+    )
+
+    assert result["outcome"] == "ok"
+    assert result["output"]["text"] == "hello world"
+
+
 def test_truthy_routes_truthy_and_falsey_outcomes() -> None:
     registry = build_registry(truthy)
     ctx = RuntimeContext(current_node_id="truthy")

tests/wf_mcp/test_service.py

@@ -273,6 +273,16 @@ def test_service_sources_have_visibility_and_capability_buckets() -> None:
     assert not std_source.capabilities.tools
 
 
+def test_wf_recipes_source_contains_composed_capabilities() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "recipes_source_store"))
+    specs = service.capability_sources["wf.recipes"].capabilities.node_specs
+
+    assert set(specs) == {"wf.recipes.extract_text_content"}
+    assert (
+        service.capability_sources["wf.recipes"].permissions.safe_for_workflow is True
+    )
+
+
 def test_wf_admin_source_exists_but_is_not_planner_visible() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "admin_source_store"))
     source = service.capability_sources["wf.admin"]
@@ -298,6 +308,7 @@ def test_service_can_disable_builtin_stdlib_specs() -> None:
     )
 
     assert "wf.std" not in service.capability_sources
+    assert "wf.recipes" not in service.capability_sources
 
 
 def test_service_planner_catalog_excludes_hidden_sources() -> None: