Progressive Discovery ahh

docs/wf_mcp_capability_sources.md

@@ -267,13 +267,12 @@ the same admin/control capabilities.
 
 ## Current Inventory Surface
 
-`list_sources()` is the source inventory:
+`list_sources()` is the compact source-discovery surface:
 
-- returns every source
+- returns paged source summaries with visibility, permissions, and counts
-- includes visibility, permissions, counts, and owned capability names
+- is intentionally compact enough for progressive discovery
-- lets callers answer planner questions by inspecting
+- pairs with `inspect_source(source_id)` for the full owned-capability inventory
-  `visibility.planner` plus `capabilities.node_specs`
 
-Humans and LLM authoring clients should use that one inventory when deciding
+Humans and LLM authoring clients should list sources first, then inspect only the
-what exists. Planner projection remains a different **use** of source metadata,
+sources they need. Planner projection remains a different **use** of source
-not a second source model.
+metadata, not a second source model.

docs/workflow_capabilities.md

@@ -251,7 +251,8 @@ The workflow-facing MCP surface now has dedicated discovery tools for the
 authoring loop:
 
 - `wf.workflow.list_capabilities`
-  - lists enabled planner-visible workflow-ready node specs
+  - lists compact paged enabled planner-visible workflow-ready node spec
+    summaries, with optional query/source filtering
 - `wf.workflow.inspect_capability`
   - returns one full workflow capability contract with schemas and outcomes
 - `wf.workflow.call_capability`

src/wf_mcp/admin_surface/handlers/broker.py

@@ -49,8 +49,16 @@ class BrokerAdminHandlers:
     def get_planner_catalog(self) -> dict[str, Any]:
         return self.service.get_planner_catalog().as_payload()
 
-    def list_sources(self) -> list[dict[str, Any]]:
+    def list_sources(
-        return self.service.list_sources()
+        self,
+        *,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        return self.service.list_source_summaries(cursor=cursor, limit=limit)
+
+    def inspect_source(self, source_id: str) -> dict[str, Any]:
+        return self.service.inspect_source(source_id)
 
     async def read_broker_resource(self, qualified_name: str) -> dict[str, Any]:
         return await self.service.read_resource(qualified_name)

src/wf_mcp/admin_surface/tools.py

@@ -82,10 +82,21 @@ def register_service_admin_tools(
     @server.tool(
         name=name("list_sources"),
         title="List Sources",
-        description="List configured capability sources and what each source owns.",
+        description="List compact configured capability source summaries.",
     )
-    async def list_sources() -> list[dict[str, Any]]:
+    async def list_sources(
-        return handlers.list_sources()
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        return handlers.list_sources(cursor=cursor, limit=limit)
+
+    @server.tool(
+        name=name("inspect_source"),
+        title="Inspect Source",
+        description="Return the full inventory for one configured capability source.",
+    )
+    async def inspect_source(source_id: str) -> dict[str, Any]:
+        return handlers.inspect_source(source_id)
 
     @server.tool(
         name=name("read_resource"),

src/wf_mcp/broker/service/core.py

@@ -23,6 +23,7 @@ from wf_platform import (
     CapabilitySource,
     SourcePermissions,
     SourceVisibility,
+    page_items,
 )
 from ...connections import ConnectionRegistry, parse_connection_id, qualify_node_name
 from ...events import EventBus, McpEvent, make_event
@@ -220,6 +221,35 @@ class WfMcpService:
             )
         ]
 
+    def list_source_summaries(
+        self,
+        *,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        """Return compact paged source summaries for progressive discovery."""
+        summaries = [
+            source.as_status().model_dump(mode="json")
+            for source in sorted(
+                self.capability_sources.values(),
+                key=lambda source: source.id,
+            )
+        ]
+        page = page_items(summaries, cursor=cursor, limit=limit)
+        return {
+            "sources": list(page.items),
+            "next_cursor": page.next_cursor,
+            "total": page.total,
+        }
+
+    def inspect_source(self, source_id: str) -> dict[str, Any]:
+        """Return the full source inventory for one exact source id."""
+        try:
+            source = self.capability_sources[source_id]
+        except KeyError as exc:
+            raise KeyError(f"unknown source {source_id!r}") from exc
+        return source.as_inventory().model_dump(mode="json")
+
     def list_available_specs(self) -> list[CatalogNodeEntry]:
         """Return planner-visible node catalog entries from every visible source."""
         return self.get_planner_catalog().entries()

src/wf_mcp/workflow_surface/handlers.py

@@ -16,7 +16,7 @@ from wf_artifacts import (
     create_workflow_artifact_from_plan as build_workflow_artifact_from_plan,
     validate_deployment_dependencies,
 )
-from wf_platform import CapabilityRef, NodeSpecInventory, hash_json_schema
+from wf_platform import CapabilityRef, NodeSpecInventory, hash_json_schema, page_items
 from wf_authoring import build_async_registry
 from wf_core import RuntimeContext
 
@@ -44,18 +44,42 @@ class WorkflowSurfaceHandlers:
         ]
         return {"nodes": entries}
 
-    async def list_capabilities(self) -> dict[str, Any]:
+    async def list_capabilities(
-        """Return planner-visible workflow-ready node spec contracts."""
+        self,
+        *,
+        query: str | None = None,
+        source_id: str | None = None,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        """Return compact paged planner-visible workflow capability summaries."""
         capabilities = [
-            detail.model_dump(mode="json")
+            {
+                "name": detail.name,
+                "description": detail.description,
+                "outcomes": list(detail.outcomes),
+                "is_async": detail.is_async,
+            }
             for source in sorted(
                 self.service.capability_sources.values(),
                 key=lambda source: source.id,
             )
             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
+            or query.casefold() in detail.name.casefold()
+            or (
+                detail.description is not None
+                and query.casefold() in detail.description.casefold()
+            )
         ]
-        return {"capabilities": capabilities}
+        page = page_items(capabilities, cursor=cursor, limit=limit)
+        return {
+            "capabilities": list(page.items),
+            "next_cursor": page.next_cursor,
+            "total": page.total,
+        }
 
     async def inspect_capability(self, *, qualified_name: str) -> dict[str, Any]:
         """Return one planner-visible workflow capability contract."""

src/wf_mcp/workflow_surface/tools.py

@@ -27,10 +27,20 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
     @server.tool(
         name="wf.workflow.list_capabilities",
         title="List Workflow Capabilities",
-        description="List planner-visible workflow-ready node capabilities.",
+        description="List compact planner-visible workflow-ready node capabilities.",
     )
-    async def list_capabilities() -> dict[str, Any]:
+    async def list_capabilities(
-        return await handlers.list_capabilities()
+        query: str | None = None,
+        source_id: str | None = None,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        return await handlers.list_capabilities(
+            query=query,
+            source_id=source_id,
+            cursor=cursor,
+            limit=limit,
+        )
 
     @server.tool(
         name="wf.workflow.inspect_capability",

src/wf_platform/__init__.py

@@ -1,4 +1,5 @@
 from .refs import CapabilityRef, SourceRef
+from .paging import Page, page_items
 from .schema_hashes import hash_json_schema
 from .sources import (
     CapabilityBuckets,
@@ -20,6 +21,7 @@ __all__ = [
     "CapabilitySource",
     "CapabilityRef",
     "NodeSpecInventory",
+    "Page",
     "ReducerInventory",
     "SourceCapabilityInventory",
     "SourceInventory",
@@ -31,4 +33,5 @@ __all__ = [
     "SourceVisibilitySnapshot",
     "SourceRef",
     "hash_json_schema",
+    "page_items",
 ]

src/wf_platform/paging.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+ItemT = TypeVar("ItemT")
+
+
+@dataclass(frozen=True, slots=True)
+class Page(Generic[ItemT]):
+    """One offset-cursor page over an already ordered in-memory sequence."""
+
+    items: tuple[ItemT, ...]
+    next_cursor: str | None
+    total: int
+
+
+def page_items(
+    items: Sequence[ItemT],
+    *,
+    cursor: str | None = None,
+    limit: int = 50,
+) -> Page[ItemT]:
+    """Return one deterministic page using a simple offset cursor."""
+    if limit < 1:
+        raise ValueError("limit must be >= 1")
+    start = 0 if cursor is None else int(cursor)
+    if start < 0:
+        raise ValueError("cursor must be >= 0")
+    end = start + limit
+    total = len(items)
+    next_cursor = str(end) if end < total else None
+    return Page(items=tuple(items[start:end]), next_cursor=next_cursor, total=total)

tests/platform/test_paging.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from wf_platform import page_items
+
+
+def test_page_items_returns_total_and_next_cursor() -> None:
+    page = page_items(["a", "b", "c"], limit=2)
+
+    assert page.items == ("a", "b")
+    assert page.next_cursor == "2"
+    assert page.total == 3
+
+
+def test_page_items_starts_from_cursor() -> None:
+    page = page_items(["a", "b", "c"], cursor="2", limit=2)
+
+    assert page.items == ("c",)
+    assert page.next_cursor is None

tests/wf_mcp/test_broker_server.py

@@ -96,7 +96,7 @@ def test_create_broker_server_exposes_tools_resources_and_prompts() -> None:
         server.call_tool("list_sources", {})
     )
     all_sources_payload = cast(dict[str, Any], cast(object, all_sources_payload_raw))
-    all_sources = all_sources_payload["result"]
+    all_sources = all_sources_payload["sources"]
     all_source_ids = {source["id"] for source in all_sources}
     assert "wf.admin" in all_source_ids
     assert "demo.personal" in all_source_ids

tests/wf_mcp/test_server.py

@@ -45,6 +45,7 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert "wf.admin.get_catalog" in names
             assert "wf.admin.get_planner_catalog" in names
             assert "wf.admin.list_sources" in names
+            assert "wf.admin.inspect_source" in names
             assert "wf.admin.read_resource" in names
             assert "wf.admin.render_prompt" in names
             assert "wf.admin.invoke_method" in names
@@ -75,7 +76,8 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert _structured(capability_result)["outcome"] == "ok"
             assert _structured(capability_result)["output"] == {"value": "hello"}
             source_ids = {
-                source["id"] for source in _structured(sources_result)["result"]
+                source["id"]
+                for source in _structured(sources_result)["sources"]
             }
             assert "wf.admin" in source_ids
             assert "wf.mcp" in source_ids

tests/wf_mcp/test_service.py

@@ -130,6 +130,17 @@ def test_service_lists_all_capability_sources_with_owned_capability_names() -> N
     assert "wf.admin.list_sources" in admin_source["capabilities"]["tools"]
 
 
+def test_service_lists_compact_source_summaries() -> None:
+    service = WfMcpService(store=FileStore(local_temp_root() / "source_summaries"))
+
+    payload = service.list_source_summaries(limit=2)
+
+    assert len(payload["sources"]) == 2
+    assert payload["total"] >= 2
+    assert payload["next_cursor"] == "2"
+    assert "capabilities" not in payload["sources"][0]
+
+
 def test_wf_std_source_contains_authoring_ops() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "stdlib_source_store"))
     specs = service.capability_sources["wf.std"].capabilities.node_specs

tests/wf_mcp/test_workflow_surface.py

@@ -77,14 +77,29 @@ def test_workflow_surface_lists_artifact_catalog_entries() -> None:
 def test_workflow_surface_lists_planner_visible_capabilities() -> None:
     handlers = _handlers(FileWorkflowArtifactStore(local_temp_root() / "surface_caps"))
 
-    payload = asyncio.run(handlers.list_capabilities())
+    payload = asyncio.run(handlers.list_capabilities(limit=2))
     names = [capability["name"] for capability in payload["capabilities"]]
 
-    assert "wf.std.runtime_error" in names
+    assert len(names) == 2
-    assert "wf.mcp.call_tool" in names
+    assert payload["total"] >= 2
+    assert payload["next_cursor"] == "2"
+    assert "description" in payload["capabilities"][0]
+    assert "input_schema" not in payload["capabilities"][0]
     assert "wf.admin.list_sources" not in names
 
 
+def test_workflow_surface_filters_capabilities_by_source() -> None:
+    handlers = _handlers(
+        FileWorkflowArtifactStore(local_temp_root() / "surface_filtered_caps")
+    )
+
+    payload = asyncio.run(handlers.list_capabilities(source_id="wf.mcp"))
+
+    assert [capability["name"] for capability in payload["capabilities"]] == [
+        "wf.mcp.call_tool"
+    ]
+
+
 def test_workflow_surface_inspects_one_capability() -> None:
     handlers = _handlers(
         FileWorkflowArtifactStore(local_temp_root() / "surface_inspect_cap")