feat: add source admin api surface

docs/current_roadmap.md

@@ -100,6 +100,10 @@ implementation state.
       public `RpcWorkflowApiClient` still satisfies `WorkflowApiSurface`, while
       client methods and server JSON-RPC registrations live in focused
       capability, draft, artifact, deployment, and run modules.
+   - Completed: read-only source inventory now has a protocol-neutral
+      `WorkflowSourceAdminApi` / `WorkflowSourceAdminSurface`; MCP admin source
+      tools delegate through it while connection/raw MCP operations remain
+      broker-owned.
 
 5. **CLI/API alignment**
    - Completed for the basic lifecycle: selected `wf` commands can target local

docs/superpowers/specs/2026-06-03-cli-api-alignment-notes.md

@@ -80,11 +80,10 @@ surface, or plain local CLI utilities.
 
 ## Next Slices
 
-1. **Server source/admin operations**
-   - Add a protocol-neutral admin/source surface for source listing, source
-     health, and dynamic source registration.
-   - Do not overload `WorkflowApiSurface` if the operation is not a workflow
-     lifecycle operation.
+1. **Source/admin transport and CLI commands**
+   - Build JSON-RPC methods and `wf source ...` commands over
+     `WorkflowSourceAdminSurface`.
+   - Keep mutation out until the store-backed source registry is designed.
 
 2. **Store-backed source registry**
    - Config can bootstrap sources, but server-owned dynamic source changes

src/wf_api/__init__.py

@@ -18,6 +18,7 @@ from .next_actions import NextActionPatchExample, NextActionTool, NextActions
 from .refs import WorkflowSurfaceCapabilityId, parse_workflow_surface_capability_id
 from .runs import WorkflowRunApi
 from .service import WorkflowApi
+from .source_admin import WorkflowSourceAdminApi
 from .surface import (
     WorkflowApiSurface,
     WorkflowArtifactSurface,
@@ -25,6 +26,7 @@ from .surface import (
     WorkflowDeploymentSurface,
     WorkflowDraftSurface,
     WorkflowRunSurface,
+    WorkflowSourceAdminSurface,
 )
 from .wrapper_hints import (
     MissingDecision,
@@ -87,6 +89,8 @@ __all__ = [
     "WorkflowRuntimeRunner",
     "WorkflowRunApi",
     "WorkflowRunSurface",
+    "WorkflowSourceAdminApi",
+    "WorkflowSourceAdminSurface",
     "WorkflowSpecProvider",
     "WorkflowSurfaceCapabilityId",
     "WrapperAuthoringHints",

src/wf_api/source_admin.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import Any
+
+from wf_platform import page_items
+
+from .operation_context import WorkflowOperationContext
+
+
+class WorkflowSourceAdminApi:
+    """Read-only protocol-neutral source inventory operations.
+
+    This is a sibling to WorkflowApi, not part of WorkflowApiSurface, because
+    source administration is server/platform management rather than workflow
+    lifecycle execution.
+    """
+
+    def __init__(self, context: WorkflowOperationContext) -> None:
+        self.context = context
+
+    async def list_sources(
+        self,
+        *,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        summaries = [
+            source.as_status().model_dump(mode="json")
+            for source in sorted(
+                self.context.specs.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,
+        }
+
+    async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
+        try:
+            source = self.context.specs.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")

src/wf_api/surface.py

@@ -199,6 +199,23 @@ class WorkflowApiSurface(
     """Public workflow operation surface shared by local and remote adapters."""
 
 
+class WorkflowSourceAdminSurface(Protocol):
+    """Read-only source/admin methods exposed by platform frontends."""
+
+    async def list_sources(
+        self,
+        *,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]: ...
+
+    async def inspect_source(
+        self,
+        *,
+        source_id: str,
+    ) -> dict[str, Any]: ...
+
+
 __all__ = [
     "WorkflowApiSurface",
     "WorkflowArtifactSurface",
@@ -206,4 +223,5 @@ __all__ = [
     "WorkflowDeploymentSurface",
     "WorkflowDraftSurface",
     "WorkflowRunSurface",
+    "WorkflowSourceAdminSurface",
 ]

src/wf_mcp/admin_surface/handlers/broker.py

@@ -1,7 +1,9 @@
 from dataclasses import asdict
 from typing import Any
 
+from wf_api import WorkflowSourceAdminApi, WorkflowSourceAdminSurface
 from wf_mcp.broker.service import WfMcpService
+from wf_mcp.broker.service.workflow_operation_context import context_from_service
 from wf_mcp.shared.errors import error_payload
 
 
@@ -10,6 +12,9 @@ class BrokerAdminHandlers:
 
     def __init__(self, service: WfMcpService) -> None:
         self.service = service
+        self.sources: WorkflowSourceAdminSurface = WorkflowSourceAdminApi(
+            context_from_service(service)
+        )
 
     def list_connections(self) -> list[dict[str, Any]]:
         return [
@@ -49,16 +54,16 @@ class BrokerAdminHandlers:
     def get_planner_catalog(self) -> dict[str, Any]:
         return self.service.get_planner_catalog().as_payload()
 
-    def list_sources(
+    async def list_sources(
         self,
         *,
         cursor: str | None = None,
         limit: int = 50,
     ) -> dict[str, Any]:
-        return self.service.list_source_summaries(cursor=cursor, limit=limit)
+        return await self.sources.list_sources(cursor=cursor, limit=limit)
 
-    def inspect_source(self, source_id: str) -> dict[str, Any]:
-        return self.service.inspect_source(source_id)
+    async def inspect_source(self, source_id: str) -> dict[str, Any]:
+        return await self.sources.inspect_source(source_id=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

@@ -107,7 +107,7 @@ def register_service_admin_tools(
             ),
         ] = 50,
     ) -> dict[str, Any]:
-        return handlers.list_sources(cursor=cursor, limit=limit)
+        return await handlers.list_sources(cursor=cursor, limit=limit)
 
     @server.tool(
         name=name("inspect_source"),
@@ -125,7 +125,7 @@ def register_service_admin_tools(
             ),
         ],
     ) -> dict[str, Any]:
-        return handlers.inspect_source(source_id)
+        return await handlers.inspect_source(source_id)
 
     @server.tool(
         name=name("read_resource"),

tests/wf_api/test_source_admin_api.py

@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+
+from wf_api import WorkflowSourceAdminApi, WorkflowSourceAdminSurface
+from wf_api.models import RawWorkflowPlan
+from wf_api.operation_context import WorkflowOperationContext
+from wf_api.saved_subgraphs import SavedSubgraphTree
+from wf_artifacts import WorkflowArtifact, WorkflowDeployment
+from wf_authoring import NodeSpec
+from wf_core import RunState
+from wf_platform import (
+    CapabilityBuckets,
+    CapabilitySource,
+    SourcePermissions,
+    SourceVisibility,
+)
+
+
+class DummyEvents:
+    def record_event(self, event: object) -> None:
+        pass
+
+    def record_workflow_event(
+        self,
+        event_type: str,
+        *,
+        capability_id: str,
+        payload: dict[str, Any],
+    ) -> None:
+        pass
+
+
+class DummyRuntime:
+    async def run_workflow_from_plan(
+        self,
+        plan: RawWorkflowPlan,
+        workflow_input: dict[str, Any],
+        deployment: WorkflowDeployment | None = None,
+        artifact: WorkflowArtifact | None = None,
+        saved_subgraph_tree: SavedSubgraphTree | None = None,
+    ) -> RunState:
+        raise AssertionError("source admin tests must not run workflows")
+
+    async def resume_workflow_from_plan(
+        self,
+        plan: RawWorkflowPlan,
+        run: RunState,
+        *,
+        resume_payload: dict[str, Any],
+        resume_outcome: str,
+        deployment: WorkflowDeployment | None = None,
+        artifact: WorkflowArtifact | None = None,
+        saved_subgraph_tree: SavedSubgraphTree | None = None,
+    ) -> RunState:
+        raise AssertionError("source admin tests must not resume workflows")
+
+
+class StaticSpecProvider:
+    def __init__(self, sources: dict[str, CapabilitySource]) -> None:
+        self._sources = sources
+
+    @property
+    def capability_sources(self) -> dict[str, CapabilitySource]:
+        return self._sources
+
+    def get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:
+        raise KeyError(f"unknown capability {qualified_name!r}")
+
+
+def _api(*sources: CapabilitySource) -> WorkflowSourceAdminApi:
+    provider = StaticSpecProvider({source.id: source for source in sources})
+    return WorkflowSourceAdminApi(
+        WorkflowOperationContext(
+            artifact_store=None,
+            draft_workspace_store=None,
+            run_store=None,
+            events=DummyEvents(),
+            specs=provider,
+            runtime=DummyRuntime(),
+            live_sources=None,
+        )
+    )
+
+
+def _source(source_id: str, *, enabled: bool = True) -> CapabilitySource:
+    return CapabilitySource(
+        id=source_id,
+        kind="connection",
+        enabled=enabled,
+        capabilities=CapabilityBuckets(),
+        visibility=SourceVisibility(
+            planner=True,
+            mcp_client=True,
+            admin_dashboard=True,
+        ),
+        permissions=SourcePermissions(calls_upstream=True),
+        description=f"{source_id} source",
+    )
+
+
+def test_source_admin_lists_compact_sources_in_id_order() -> None:
+    api = _api(_source("zeta.personal"), _source("alpha.personal", enabled=False))
+
+    payload = asyncio.run(api.list_sources(limit=10))
+
+    assert payload["total"] == 2
+    assert payload["next_cursor"] is None
+    assert [source["id"] for source in payload["sources"]] == [
+        "alpha.personal",
+        "zeta.personal",
+    ]
+    assert payload["sources"][0]["enabled"] is False
+    assert payload["sources"][1]["description"] == "zeta.personal source"
+
+
+def test_source_admin_pages_sources() -> None:
+    api = _api(_source("a"), _source("b"), _source("c"))
+
+    first = asyncio.run(api.list_sources(limit=2))
+    second = asyncio.run(api.list_sources(cursor=first["next_cursor"], limit=2))
+
+    assert [source["id"] for source in first["sources"]] == ["a", "b"]
+    assert first["next_cursor"] == "2"
+    assert [source["id"] for source in second["sources"]] == ["c"]
+    assert second["next_cursor"] is None
+
+
+def test_source_admin_inspects_full_source_inventory() -> None:
+    api = _api(_source("demo.personal"))
+
+    payload = asyncio.run(api.inspect_source(source_id="demo.personal"))
+
+    assert payload["id"] == "demo.personal"
+    assert payload["kind"] == "connection"
+    assert payload["description"] == "demo.personal source"
+    assert payload["visibility"]["planner"] is True
+    assert payload["permissions"]["calls_upstream"] is True
+
+
+def test_source_admin_inspect_unknown_source_raises_clear_key_error() -> None:
+    api = _api(_source("demo.personal"))
+
+    with pytest.raises(KeyError, match="unknown source 'missing.source'"):
+        asyncio.run(api.inspect_source(source_id="missing.source"))
+
+
+def test_source_admin_api_satisfies_surface_protocol() -> None:
+    api: WorkflowSourceAdminSurface = _api(_source("demo.personal"))
+
+    assert api is not None

tests/wf_mcp/test_admin_surface.py

@@ -26,6 +26,13 @@ def test_broker_admin_handlers_list_connections_and_events() -> None:
     assert events[0]["kind"] == "connection_registered"
     assert events[0]["connection_id"] == "demo.personal"
 
+    sources = _run(handlers.list_sources(limit=100))
+
+    source_ids = {source["id"] for source in sources["sources"]}
+    assert "wf.std" in source_ids
+    assert "wf.admin" in source_ids
+    assert sources["total"] >= 2
+
 
 def test_broker_admin_handlers_report_failed_refresh_payload() -> None:
     service = WfMcpService(store=FileStore(local_temp_root() / "admin_refresh_store"))