list_something pattern generalized

src/wf_mcp/shared/__init__.py

@@ -1,4 +1,5 @@
 from .errors import error_payload, root_exception
+from .listing import matches_query, paged_list_payload
 from .names import (
     ADMIN_NAMESPACE,
     LdaNamespace,
@@ -21,7 +22,9 @@ __all__ = [
     "error_payload",
     "is_admin_tool_name",
     "make_cursor",
+    "matches_query",
     "namespaced_tool_name",
+    "paged_list_payload",
     "paginate_items",
     "parse_cursor",
     "parse_namespaced_tool_name",

src/wf_mcp/shared/listing.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any, TypeVar
+
+from wf_platform import page_items
+
+T = TypeVar("T")
+
+
+def matches_query(*values: object, query: str | None) -> bool:
+    """Return whether a compact discovery row matches a human search query.
+
+    List tools intentionally search across a few summary/display fields rather
+    than returning every detail. Inspect tools remain the path for full schemas.
+    """
+    if query is None:
+        return True
+    needle = query.strip().casefold()
+    if not needle:
+        return True
+    return any(needle in str(value).casefold() for value in values if value is not None)
+
+
+def paged_list_payload(
+    key: str,
+    items: Sequence[T],
+    *,
+    cursor: str | None,
+    limit: int,
+) -> dict[str, Any]:
+    """Build the common workflow-surface list response shape.
+
+    Workflow admin tools use simple offset cursors (`"50"`, `"100"`) because
+    these payloads are tool results, not raw MCP protocol `tools/list` pages.
+    """
+    page = page_items(items, cursor=cursor, limit=limit)
+    return {
+        key: list(page.items),
+        "next_cursor": page.next_cursor,
+        "total": page.total,
+    }

src/wf_mcp/workflow_surface/handlers.py

@@ -28,7 +28,6 @@ from wf_platform import (
     CapabilitySource,
     NodeSpecInventory,
     hash_json_schema,
-    page_items,
 )
 from wf_authoring import build_async_registry
 from wf_core import RuntimeContext
@@ -42,6 +41,7 @@ from wf_core.paths import GraphSourcePath, LocalPath, StatePath
 
 from ..events import make_event
 from ..models import RawWorkflowPlan
+from ..shared import matches_query, paged_list_payload
 from .constants import (
     DEFAULT_CALL_STEP_ID,
     DEFAULT_ERROR_OUTCOME,
@@ -68,16 +68,42 @@ class WorkflowSurfaceHandlers:
     def __init__(self, service: WfMcpService) -> None:
         self.service = service
 
-    async def list_artifacts(self) -> dict[str, Any]:
+    async def list_artifacts(
+        self,
+        *,
+        query: str | None = None,
+        kind: ArtifactKind | None = None,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        """Return compact paged saved artifact summaries.
+
+        Saved artifacts can contain full raw workflow plans, so list results
+        deliberately stay summary-only. Use inspect/run tools for detail.
+        """
         if self.service.artifact_store is None:
-            return {"nodes": []}
+            return paged_list_payload("nodes", [], cursor=cursor, limit=limit)
         entries = [
             self.service.workflow_artifact_catalog_entry(artifact).model_dump(
                 mode="json"
             )
             for artifact in self.service.artifact_store.list_artifacts()
+            if kind is None or artifact.kind == kind
         ]
-        return {"nodes": entries}
+        entries = [
+            entry
+            for entry in entries
+            if matches_query(
+                entry.get("name"),
+                entry.get("artifact_id"),
+                entry.get("display_name"),
+                entry.get("description"),
+                entry.get("kind"),
+                query=query,
+            )
+        ]
+        entries.sort(key=lambda entry: str(entry["name"]))
+        return paged_list_payload("nodes", entries, cursor=cursor, limit=limit)
 
     async def list_capabilities(
         self,
@@ -106,23 +132,22 @@ class WorkflowSurfaceHandlers:
             if source.enabled and source.visibility.planner
             if source_id is None or source.id == source_id
             for detail in source.as_inventory().capabilities.node_spec_details
-            if query is None
+            if matches_query(
-            or query.casefold() in detail.name.casefold()
+                detail.name,
-            or (
+                detail.description,
-                detail.description is not None
+                query=query,
-                and query.casefold() in detail.description.casefold()
             )
         ]
         capabilities.extend(
             self._wrapper_capability_summaries(query=query, source_id=source_id)
         )
         capabilities.sort(key=lambda capability: capability["name"])
-        page = page_items(capabilities, cursor=cursor, limit=limit)
+        return paged_list_payload(
-        return {
+            "capabilities",
-            "capabilities": list(page.items),
+            capabilities,
-            "next_cursor": page.next_cursor,
+            cursor=cursor,
-            "total": page.total,
+            limit=limit,
-        }
+        )
 
     async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
         """Return one planner-visible workflow capability contract."""
@@ -221,7 +246,7 @@ class WorkflowSurfaceHandlers:
             if artifact.kind != "wrapper":
                 continue
             name = _artifact_capability_id(artifact)
-            if not _matches_capability_query(
+            if not matches_query(
                 name,
                 artifact.description,
                 query=query,
@@ -1063,21 +1088,6 @@ def _schema_field_names(schema: dict[str, Any]) -> list[str]:
     return sorted(str(name) for name in properties)
 
 
-def _matches_capability_query(
-    name: str,
-    description: str | None,
-    *,
-    query: str | None,
-) -> bool:
-    """Apply the same compact capability search semantics to every row kind."""
-    if query is None:
-        return True
-    lowered = query.casefold()
-    return lowered in name.casefold() or (
-        description is not None and lowered in description.casefold()
-    )
-
-
 def _first_state_path(output_map: dict[str, str]) -> str | None:
     """Return the first mapped state path for minimal error-route bootstraps."""
     for target in output_map.values():

src/wf_mcp/workflow_surface/tools.py

@@ -39,10 +39,43 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
     @server.tool(
         name="wf.workflow.list_artifacts",
         title="List Workflow Artifacts",
-        description="List saved workflow artifacts available to run or inspect.",
+        description=(
+            "List compact saved workflow artifacts. Use query/kind for "
+            "discovery, then inspect or run a selected artifact for detail."
+        ),
     )
-    async def list_artifacts() -> dict[str, Any]:
+    async def list_artifacts(
-        return await handlers.list_artifacts()
+        query: Annotated[
+            str | None,
+            Field(
+                description=(
+                    "Optional case-insensitive search across artifact name, id, "
+                    "display name, description, and kind."
+                )
+            ),
+        ] = None,
+        kind: Annotated[
+            ArtifactKind | None,
+            Field(description="Optional artifact kind filter: workflow or wrapper."),
+        ] = None,
+        cursor: Annotated[
+            str | None,
+            Field(description="Opaque offset cursor returned by a previous page."),
+        ] = None,
+        limit: Annotated[
+            int,
+            Field(
+                ge=1,
+                description="Maximum artifact summaries to return in this page.",
+            ),
+        ] = 50,
+    ) -> dict[str, Any]:
+        return await handlers.list_artifacts(
+            query=query,
+            kind=kind,
+            cursor=cursor,
+            limit=limit,
+        )
 
     @server.tool(
         name="wf.workflow.list_capabilities",
@@ -54,10 +87,30 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
         ),
     )
     async def list_capabilities(
-        query: str | None = None,
+        query: Annotated[
-        source_id: str | None = None,
+            str | None,
-        cursor: str | None = None,
+            Field(
-        limit: int = 50,
+                description=(
+                    "Optional case-insensitive search across capability name and "
+                    "description."
+                )
+            ),
+        ] = None,
+        source_id: Annotated[
+            str | None,
+            Field(description="Optional capability source id filter, such as wf.std."),
+        ] = None,
+        cursor: Annotated[
+            str | None,
+            Field(description="Opaque offset cursor returned by a previous page."),
+        ] = None,
+        limit: Annotated[
+            int,
+            Field(
+                ge=1,
+                description="Maximum capability summaries to return in this page.",
+            ),
+        ] = 50,
     ) -> dict[str, Any]:
         return await handlers.list_capabilities(
             query=query,

tests/wf_mcp/test_server.py

@@ -193,6 +193,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
 
             assert _structured(echo_result)["echoed"] == "hello"
             assert _structured(artifacts_result)["nodes"] == []
+            assert _structured(artifacts_result)["total"] == 0
             assert _structured(capability_result)["outcome"] == "ok"
             assert _structured(capability_result)["output"] == {"value": "hello"}
             source_ids = {
@@ -423,6 +424,7 @@ def test_server_safe_tool_names_adapts_dotted_runtime_names() -> None:
             )
 
             assert artifacts["nodes"] == []
+            assert artifacts["total"] == 0
             assert echo["echoed"] == "hello"
 
     asyncio.run(run_proxy())
@@ -444,6 +446,10 @@ def test_workflow_tools_have_human_metadata() -> None:
 
             assert list_artifacts.title == "List Workflow Artifacts"
             assert "saved workflow artifacts" in (list_artifacts.description or "")
+            assert "query" in list_artifacts.inputSchema["properties"]
+            assert "kind" in list_artifacts.inputSchema["properties"]
+            assert "cursor" in list_artifacts.inputSchema["properties"]
+            assert "limit" in list_artifacts.inputSchema["properties"]
             assert run_deployment.title == "Run Workflow Deployment"
             assert "deployment_id" in (run_deployment.description or "")
             assert "trace_range" in run_deployment.inputSchema["properties"]

tests/wf_mcp/test_workflow_surface.py

@@ -74,6 +74,8 @@ def test_workflow_surface_lists_artifact_catalog_entries() -> None:
 
     nodes = payload["nodes"]
     assert len(nodes) == 1
+    assert payload["total"] == 1
+    assert payload["next_cursor"] is None
     assert nodes[0]["name"] == "workflow.summarize_docs.v1"
     assert nodes[0]["artifact_id"] == "summarize_docs"
     assert nodes[0]["version"] == 1
@@ -82,6 +84,50 @@ def test_workflow_surface_lists_artifact_catalog_entries() -> None:
     assert "plan" not in nodes[0]
 
 
+def test_workflow_surface_pages_and_filters_artifact_catalog_entries() -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_artifact_pages"
+    )
+    artifact_store.save_artifact(_artifact())
+    artifact_store.save_artifact(
+        _artifact().model_copy(
+            update={
+                "id": "echo_wrapper",
+                "version": 2,
+                "kind": "wrapper",
+                "title": "Echo Wrapper",
+                "description": "Reusable echo wrapper.",
+            }
+        )
+    )
+    artifact_store.save_artifact(
+        _artifact().model_copy(
+            update={
+                "id": "browser_click",
+                "title": "Browser Click",
+                "description": "Open a page and wait for a click.",
+            }
+        )
+    )
+    handlers = _handlers(artifact_store)
+
+    first_page = asyncio.run(handlers.list_artifacts(limit=2))
+    second_page = asyncio.run(
+        handlers.list_artifacts(cursor=first_page["next_cursor"], limit=2)
+    )
+    wrappers = asyncio.run(handlers.list_artifacts(kind="wrapper", query="echo"))
+
+    assert first_page["total"] == 3
+    assert first_page["next_cursor"] == "2"
+    assert len(first_page["nodes"]) == 2
+    assert len(second_page["nodes"]) == 1
+    assert second_page["next_cursor"] is None
+    assert wrappers["total"] == 1
+    assert wrappers["nodes"][0]["artifact_id"] == "echo_wrapper"
+    assert wrappers["nodes"][0]["kind"] == "wrapper"
+    assert "plan" not in wrappers["nodes"][0]
+
+
 def test_workflow_surface_lists_planner_visible_capabilities() -> None:
     handlers = _handlers(FileWorkflowArtifactStore(local_temp_root() / "surface_caps"))