less dicts more models

docs/wf_mcp_capability_sources.md

@@ -191,6 +191,9 @@ The code now has the first capability-source layer in place.
 
 - `wf_platform.CapabilitySource` owns source metadata, visibility, permissions, and
   capability buckets.
+- `CapabilitySource` is the mutable runtime registry object; typed
+  `SourceStatus` and `SourceInventory` snapshots are the serializable domain
+  projections used at boundaries such as `list_sources()`.
 - `WfMcpService.capability_sources` is the canonical in-memory registry.
 - Planner node lookup reads `CapabilitySource.capabilities.node_specs`
   directly; the old `SpecSource` compatibility layer has been removed.

src/wf_mcp/broker/service/core.py

@@ -213,7 +213,7 @@ class WfMcpService:
     def list_sources(self) -> list[dict[str, Any]]:
         """Return every capability source with the names it currently owns."""
         return [
-            source.as_inventory()
+            source.as_inventory().model_dump(mode="json")
             for source in sorted(
                 self.capability_sources.values(),
                 key=lambda source: source.id,

src/wf_platform/__init__.py

@@ -1,15 +1,25 @@
 from .sources import (
     CapabilityBuckets,
     CapabilitySource,
+    SourceCapabilityInventory,
+    SourceInventory,
     SourceKind,
     SourcePermissions,
+    SourcePermissionsSnapshot,
+    SourceStatus,
     SourceVisibility,
+    SourceVisibilitySnapshot,
 )
 
 __all__ = [
     "CapabilityBuckets",
     "CapabilitySource",
+    "SourceCapabilityInventory",
+    "SourceInventory",
     "SourceKind",
     "SourcePermissions",
+    "SourcePermissionsSnapshot",
+    "SourceStatus",
     "SourceVisibility",
+    "SourceVisibilitySnapshot",
 ]

src/wf_platform/sources.py

@@ -1,6 +1,8 @@
 from dataclasses import dataclass, field
 from typing import Any, Literal
 
+from pydantic import BaseModel
+
 from wf_authoring import NodeSpec
 from wf_core import ReducerSpec
 from wf_core.runtime.ops.merges import ReducerDefinition
@@ -23,6 +25,55 @@ class SourcePermissions:
     mutates_auth: bool = False
 
 
+class SourceVisibilitySnapshot(BaseModel):
+    """Serializable visibility flags for one source inventory snapshot."""
+
+    planner: bool = False
+    mcp_client: bool = False
+    admin_dashboard: bool = True
+
+
+class SourcePermissionsSnapshot(BaseModel):
+    """Serializable permission flags for one source inventory snapshot."""
+
+    safe_for_workflow: bool = False
+    calls_upstream: bool = False
+    mutates_config: bool = False
+    mutates_auth: bool = False
+
+
+class SourceCapabilityInventory(BaseModel):
+    """Serializable names owned by one source, grouped by capability kind."""
+
+    tools: tuple[str, ...] = ()
+    node_specs: tuple[str, ...] = ()
+    reducers: tuple[str, ...] = ()
+    prompts: tuple[str, ...] = ()
+    resources: tuple[str, ...] = ()
+
+
+class SourceStatus(BaseModel):
+    """Serializable source metadata without the full owned-name inventory."""
+
+    id: str
+    kind: SourceKind
+    enabled: bool
+    visibility: SourceVisibilitySnapshot
+    permissions: SourcePermissionsSnapshot
+    description: str | None = None
+    tool_count: int
+    node_spec_count: int
+    reducer_count: int
+    prompt_count: int
+    resource_count: int
+
+
+class SourceInventory(SourceStatus):
+    """Serializable source snapshot with the capability names it owns."""
+
+    capabilities: SourceCapabilityInventory
+
+
 @dataclass(slots=True)
 class CapabilityBuckets:
     tools: dict[str, Any] = field(default_factory=dict)
@@ -43,39 +94,40 @@ class CapabilitySource:
     permissions: SourcePermissions = field(default_factory=SourcePermissions)
     description: str | None = None
 
-    def as_status(self) -> dict[str, Any]:
+    def as_status(self) -> SourceStatus:
-        return {
+        """Return serializable source metadata without owned capability names."""
-            "id": self.id,
+        return SourceStatus(
-            "kind": self.kind,
+            id=self.id,
-            "enabled": self.enabled,
+            kind=self.kind,
-            "visibility": {
+            enabled=self.enabled,
-                "planner": self.visibility.planner,
+            visibility=SourceVisibilitySnapshot(
-                "mcp_client": self.visibility.mcp_client,
+                planner=self.visibility.planner,
-                "admin_dashboard": self.visibility.admin_dashboard,
+                mcp_client=self.visibility.mcp_client,
-            },
+                admin_dashboard=self.visibility.admin_dashboard,
-            "permissions": {
+            ),
-                "safe_for_workflow": self.permissions.safe_for_workflow,
+            permissions=SourcePermissionsSnapshot(
-                "calls_upstream": self.permissions.calls_upstream,
+                safe_for_workflow=self.permissions.safe_for_workflow,
-                "mutates_config": self.permissions.mutates_config,
+                calls_upstream=self.permissions.calls_upstream,
-                "mutates_auth": self.permissions.mutates_auth,
+                mutates_config=self.permissions.mutates_config,
-            },
+                mutates_auth=self.permissions.mutates_auth,
-            "description": self.description,
+            ),
-            "tool_count": len(self.capabilities.tools),
+            description=self.description,
-            "node_spec_count": len(self.capabilities.node_specs),
+            tool_count=len(self.capabilities.tools),
-            "reducer_count": len(self.capabilities.reducers),
+            node_spec_count=len(self.capabilities.node_specs),
-            "prompt_count": len(self.capabilities.prompts),
+            reducer_count=len(self.capabilities.reducers),
-            "resource_count": len(self.capabilities.resources),
+            prompt_count=len(self.capabilities.prompts),
-        }
+            resource_count=len(self.capabilities.resources),
+        )
 
-    def as_inventory(self) -> dict[str, Any]:
+    def as_inventory(self) -> SourceInventory:
-        """Return source metadata plus the capability names it owns."""
+        """Return a serializable source snapshot plus owned capability names."""
-        return {
+        return SourceInventory(
-            **self.as_status(),
+            **self.as_status().model_dump(),
-            "capabilities": {
+            capabilities=SourceCapabilityInventory(
-                "tools": sorted(self.capabilities.tools),
+                tools=tuple(sorted(self.capabilities.tools)),
-                "node_specs": sorted(self.capabilities.node_specs),
+                node_specs=tuple(sorted(self.capabilities.node_specs)),
-                "reducers": sorted(self.capabilities.reducers),
+                reducers=tuple(sorted(self.capabilities.reducers)),
-                "prompts": sorted(self.capabilities.prompts),
+                prompts=tuple(sorted(self.capabilities.prompts)),
-                "resources": sorted(self.capabilities.resources),
+                resources=tuple(sorted(self.capabilities.resources)),
-            },
+            ),
-        }
+        )

tests/platform/test_sources.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from wf_platform import (
+    CapabilityBuckets,
+    CapabilitySource,
+    SourceInventory,
+    SourcePermissions,
+    SourceStatus,
+    SourceVisibility,
+)
+
+
+def test_capability_source_projects_typed_status() -> None:
+    source = CapabilitySource(
+        id="wf.std",
+        kind="system",
+        capabilities=CapabilityBuckets(
+            tools={"wf.std.inspect": object()},
+            resources={"wf.std.manual": object()},
+        ),
+        visibility=SourceVisibility(planner=True, mcp_client=True),
+        permissions=SourcePermissions(safe_for_workflow=True),
+        description="Workflow standard library.",
+    )
+
+    status = source.as_status()
+
+    assert isinstance(status, SourceStatus)
+    assert status.id == "wf.std"
+    assert status.visibility.planner is True
+    assert status.permissions.safe_for_workflow is True
+    assert status.tool_count == 1
+    assert status.resource_count == 1
+
+
+def test_capability_source_projects_typed_inventory() -> None:
+    source = CapabilitySource(
+        id="wf.std",
+        kind="system",
+        capabilities=CapabilityBuckets(
+            tools={"wf.std.inspect": object()},
+            resources={"wf.std.manual": object()},
+        ),
+    )
+
+    inventory = source.as_inventory()
+
+    assert isinstance(inventory, SourceInventory)
+    assert inventory.capabilities.tools == ("wf.std.inspect",)
+    assert inventory.capabilities.resources == ("wf.std.manual",)
+    assert inventory.model_dump(mode="json")["capabilities"]["tools"] == [
+        "wf.std.inspect"
+    ]