feat: add stateful runtime listings

docs/current_roadmap.md

@@ -256,6 +256,10 @@ implementation state.
     - Completed: broker content access now prefers a configured stateful MCP
       runtime for `read_resource` and `get_prompt`, with one-shot adapter fallback.
       Catalog refresh/discovery remains one-shot by policy.
+    - Completed: stateful MCP runtime now has protocol slices for tools,
+      resources, and prompts, and can route session-scoped `list_resources` and
+      `list_prompts` through the owner task. Catalog refresh still uses one-shot
+      adapter policy.
   - Auth/source secrets boundary: keep registry desired state separate from
     upstream credentials, and surface missing auth as validation diagnostics.
     The contract is now specified in

docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md

@@ -123,7 +123,10 @@ First slices should move leaf modules only and leave `wf_mcp` re-export shims:
 13. Complete: broker content access now prefers configured `StatefulMcpRuntime`
     for resource and prompt reads, falling back to the one-shot adapter when no
     stateful runtime is configured. Catalog refresh/discovery remains one-shot.
-14. Upstream transport/discovery/session services.
+14. Complete: stateful MCP runtime protocols split into tool/resource/prompt
+    slices. Runtime can route `list_resources` and `list_prompts` through the
+    owner task for session-scoped listings; catalog refresh remains one-shot.
+15. Upstream transport/discovery/session services.
 
 Each slice should add import-direction tests so the new source-provider package
 does not depend on `wf_mcp.workflow_surface`, `wf_mcp.admin_surface`,

src/wf_mcp/sdk/__init__.py

@@ -1,8 +1,19 @@
 from wf_sources_mcp.sdk import (
     BackendAdapter,
     McpSdkAdapter,
+    PromptRuntime,
+    ResourceRuntime,
     StatefulMcpRuntime,
     ToolCallResult,
+    ToolRuntime,
 )
 
-__all__ = ["BackendAdapter", "McpSdkAdapter", "StatefulMcpRuntime", "ToolCallResult"]
+__all__ = [
+    "BackendAdapter",
+    "McpSdkAdapter",
+    "PromptRuntime",
+    "ResourceRuntime",
+    "StatefulMcpRuntime",
+    "ToolCallResult",
+    "ToolRuntime",
+]

src/wf_mcp/sdk/base.py

@@ -5,10 +5,20 @@ Canonical implementation lives in `wf_sources_mcp.sdk`.
 
 from __future__ import annotations
 
-from wf_sources_mcp.sdk import BackendAdapter, StatefulMcpRuntime, ToolCallResult
+from wf_sources_mcp.sdk import (
+    BackendAdapter,
+    PromptRuntime,
+    ResourceRuntime,
+    StatefulMcpRuntime,
+    ToolCallResult,
+    ToolRuntime,
+)
 
 __all__ = [
     "BackendAdapter",
+    "PromptRuntime",
+    "ResourceRuntime",
     "StatefulMcpRuntime",
     "ToolCallResult",
+    "ToolRuntime",
 ]

src/wf_sources_mcp/runtime/factory.py

@@ -11,6 +11,7 @@ from typing import Any, Generic, TypeVar
 from mcp.client.session import ClientSession
 
 from wf_sources_mcp.auth import AuthRecord
+from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource
 from wf_sources_mcp.client import McpSourceClient, open_mcp_session
 from wf_sources_mcp.connections import McpSourceConnection
 from wf_sources_mcp.sdk import ToolCallResult
@@ -44,6 +45,8 @@ class PersistentSessionFactory:
             call_callback=owner.call_tool,
             read_resource_callback=owner.read_resource,
             get_prompt_callback=owner.get_prompt,
+            list_resources_callback=owner.list_resources,
+            list_prompts_callback=owner.list_prompts,
             close_callback=owner.close,
         )
 
@@ -163,6 +166,20 @@ class _SessionOwner:
             run=lambda client: client.get_prompt(prompt_name, arguments),
         )
 
+    async def list_resources(self) -> list[DiscoveredResource]:
+        """Submit resource listing through the generic owner-task operation queue."""
+        return await self.submit(
+            operation="list_resources",
+            run=lambda client: client.list_resources(),
+        )
+
+    async def list_prompts(self) -> list[DiscoveredPrompt]:
+        """Submit prompt listing through the generic owner-task operation queue."""
+        return await self.submit(
+            operation="list_prompts",
+            run=lambda client: client.list_prompts(),
+        )
+
     async def close(self) -> None:
         """Ask the owner task to close the MCP transport in its own scope."""
         task = self._task

src/wf_sources_mcp/runtime/pool.py

@@ -7,6 +7,7 @@ from inspect import isawaitable
 from typing import Any, cast
 
 from wf_sources_mcp.auth import AuthRecord
+from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource
 from wf_sources_mcp.connections import McpSourceConnection
 from wf_sources_mcp.sdk import ToolCallResult
 
@@ -102,6 +103,22 @@ class McpRuntimePool:
         session = await self.get_session(connection, auth)
         return await session.get_prompt(prompt_name, arguments)
 
+    async def list_resources(
+        self,
+        connection: McpSourceConnection,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredResource]:
+        session = await self.get_session(connection, auth)
+        return await session.list_resources()
+
+    async def list_prompts(
+        self,
+        connection: McpSourceConnection,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredPrompt]:
+        session = await self.get_session(connection, auth)
+        return await session.list_prompts()
+
     async def close_connection(self, connection_id: str) -> None:
         current = self._sessions.pop(connection_id, None)
         if current is not None:

src/wf_sources_mcp/runtime/session.py

@@ -8,6 +8,7 @@ from mcp.client.session import ClientSession
 from pydantic import AnyUrl
 
 from wf_sources_mcp.auth import AuthRecord
+from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource
 from wf_sources_mcp.connections import McpSourceConnection
 from wf_sources_mcp.sdk import ToolCallResult
 from wf_sources_mcp.sdk.converters import tool_result_to_call_result
@@ -18,20 +19,18 @@ RawPromptGetter = Callable[
     [str, dict[str, str] | None],
     Awaitable[dict[str, Any]],
 ]
+RawResourceLister = Callable[[], Awaitable[list[DiscoveredResource]]]
+RawPromptLister = Callable[[], Awaitable[list[DiscoveredPrompt]]]
 
 
 @dataclass(slots=True)
 class PersistentMcpSession:
     """Long-lived MCP execution handle for one configured connection.
 
-    Production sessions use `call_callback` because MCP transports are entered
-    inside an AnyIO cancel scope and must be called and closed by that same
-    owner task. `client` remains available for simple injected/fake sessions in
-    tests. `call_tool()` always normalizes SDK results for workflow nodes.
-
-    This runtime intentionally exposes only tool calls for now. Shared
-    non-tool operations live on `McpSourceClient`; routing them through the
-    owner task is a separate runtime-expansion slice.
+    Production sessions route all MCP operations through the owner-task queue
+    for session safety. `client` remains available for simple injected/fake
+    sessions in tests. `call_tool()` always normalizes SDK results for workflow
+    nodes.
     """
 
     connection: McpSourceConnection
@@ -40,6 +39,8 @@ class PersistentMcpSession:
     call_callback: RawToolCaller | None = None
     read_resource_callback: RawResourceReader | None = None
     get_prompt_callback: RawPromptGetter | None = None
+    list_resources_callback: RawResourceLister | None = None
+    list_prompts_callback: RawPromptLister | None = None
     close_callback: Callable[[], Awaitable[None]] | None = None
 
     async def call_tool(
@@ -74,6 +75,28 @@ class PersistentMcpSession:
             return result.model_dump(by_alias=True, mode="json", exclude_none=True)
         raise RuntimeError("persistent MCP session has no prompt transport")
 
+    async def list_resources(self) -> list[DiscoveredResource]:
+        """List MCP resources through the owner task or injected session."""
+        if self.list_resources_callback is not None:
+            return await self.list_resources_callback()
+        if self.client is not None:
+            from wf_sources_mcp.sdk.converters import resource_to_discovered
+
+            result = await self.client.list_resources()
+            return [resource_to_discovered(resource) for resource in result.resources]
+        raise RuntimeError("persistent MCP session has no resource list transport")
+
+    async def list_prompts(self) -> list[DiscoveredPrompt]:
+        """List MCP prompts through the owner task or injected session."""
+        if self.list_prompts_callback is not None:
+            return await self.list_prompts_callback()
+        if self.client is not None:
+            from wf_sources_mcp.sdk.converters import prompt_to_discovered
+
+            result = await self.client.list_prompts()
+            return [prompt_to_discovered(prompt) for prompt in result.prompts]
+        raise RuntimeError("persistent MCP session has no prompt list transport")
+
     async def close(self) -> None:
         """Close the transport/session stack owned by the runtime factory."""
         if self.close_callback is not None:

src/wf_sources_mcp/sdk/__init__.py

@@ -8,14 +8,25 @@ from .converters import (
     tool_to_discovered,
     workflow_output_schema_from_mcp_tool_schema,
 )
-from .protocols import BackendAdapter, StatefulMcpRuntime, ToolCallResult, ToolExecutor
+from .protocols import (
+    BackendAdapter,
+    PromptRuntime,
+    ResourceRuntime,
+    StatefulMcpRuntime,
+    ToolCallResult,
+    ToolExecutor,
+    ToolRuntime,
+)
 
 __all__ = [
     "BackendAdapter",
     "McpSdkAdapter",
+    "PromptRuntime",
+    "ResourceRuntime",
     "StatefulMcpRuntime",
     "ToolCallResult",
     "ToolExecutor",
+    "ToolRuntime",
     "prompt_to_discovered",
     "resource_to_discovered",
     "tool_result_to_call_result",

src/wf_sources_mcp/sdk/protocols.py

@@ -82,13 +82,8 @@ class BackendAdapter(Protocol):
     ) -> ToolCallResult: ...
 
 
-class ToolExecutor(Protocol):
-    """Runtime boundary for executing MCP tools from workflow nodes.
-
-    Discovery can stay one-shot, but workflow execution needs this smaller
-    protocol so persistent runtime pools can replace one-shot adapters without
-    changing generated NodeSpecs.
-    """
+class ToolRuntime(Protocol):
+    """Runtime boundary for executing MCP tools from workflow nodes."""
 
     async def call_tool(
         self,
@@ -99,12 +94,18 @@ class ToolExecutor(Protocol):
     ) -> ToolCallResult: ...
 
 
-class StatefulMcpRuntime(ToolExecutor, Protocol):
-    """Stateful execution/read boundary for configured MCP sources.
+class ToolExecutor(ToolRuntime, Protocol):
+    """Compatibility name for workflow-node tool execution."""
 
-    Implementations keep source session state across calls. Discovery/catalog
-    refresh may still use one-shot adapters by policy.
-    """
+
+class ResourceRuntime(Protocol):
+    """Stateful resource operations for configured MCP sources."""
+
+    async def list_resources(
+        self,
+        connection: McpSourceConnection,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredResource]: ...
 
     async def read_resource(
         self,
@@ -113,6 +114,16 @@ class StatefulMcpRuntime(ToolExecutor, Protocol):
         uri: str,
     ) -> dict[str, Any]: ...
 
+
+class PromptRuntime(Protocol):
+    """Stateful prompt operations for configured MCP sources."""
+
+    async def list_prompts(
+        self,
+        connection: McpSourceConnection,
+        auth: AuthRecord | None,
+    ) -> list[DiscoveredPrompt]: ...
+
     async def get_prompt(
         self,
         connection: McpSourceConnection,
@@ -122,9 +133,20 @@ class StatefulMcpRuntime(ToolExecutor, Protocol):
     ) -> dict[str, Any]: ...
 
 
+class StatefulMcpRuntime(ToolRuntime, ResourceRuntime, PromptRuntime, Protocol):
+    """Stateful execution/read/list boundary for configured MCP sources.
+
+    Implementations keep source session state across calls. Catalog refresh may
+    still use one-shot adapters by policy.
+    """
+
+
 __all__ = [
     "BackendAdapter",
+    "PromptRuntime",
+    "ResourceRuntime",
     "StatefulMcpRuntime",
     "ToolCallResult",
     "ToolExecutor",
+    "ToolRuntime",
 ]

tests/wf_mcp/test_compat_imports.py

@@ -120,6 +120,21 @@ def test_wf_mcp_sdk_protocol_shims_reexport_wf_sources_mcp_sdk() -> None:
     assert CompatBaseToolCallResult is ToolCallResult
     assert CompatBaseStatefulMcpRuntime is StatefulMcpRuntime
 
+    from wf_mcp.sdk import PromptRuntime as CompatPromptRuntime
+    from wf_mcp.sdk import ResourceRuntime as CompatResourceRuntime
+    from wf_mcp.sdk import ToolRuntime as CompatToolRuntime
+    from wf_mcp.sdk.base import PromptRuntime as CompatBasePromptRuntime
+    from wf_mcp.sdk.base import ResourceRuntime as CompatBaseResourceRuntime
+    from wf_mcp.sdk.base import ToolRuntime as CompatBaseToolRuntime
+    from wf_sources_mcp.sdk import PromptRuntime, ResourceRuntime, ToolRuntime
+
+    assert CompatPromptRuntime is PromptRuntime
+    assert CompatResourceRuntime is ResourceRuntime
+    assert CompatToolRuntime is ToolRuntime
+    assert CompatBasePromptRuntime is PromptRuntime
+    assert CompatBaseResourceRuntime is ResourceRuntime
+    assert CompatBaseToolRuntime is ToolRuntime
+
 
 def test_wf_mcp_runtime_protocol_shim_reexports_wf_sources_mcp_tool_executor() -> None:
     from wf_mcp.runtime import ToolExecutor as CompatRuntimeToolExecutor

tests/wf_sources_mcp/test_runtime.py

@@ -6,7 +6,13 @@ from typing import Any
 import pytest
 from mcp.client.session import ClientSession
 from mcp.types import CallToolResult as RawCallToolResult
-from mcp.types import TextContent
+from mcp.types import (
+    ListPromptsResult,
+    ListResourcesResult,
+    Prompt,
+    Resource,
+    TextContent,
+)
 from pydantic import AnyUrl
 
 from wf_sources_mcp.auth import AuthRecord
@@ -136,6 +142,31 @@ class _FakeFactory(PersistentSessionFactory):
                     },
                 )()
 
+            async def list_resources(self) -> ListResourcesResult:
+                return ListResourcesResult(
+                    resources=[
+                        Resource(
+                            uri=AnyUrl("fixture://docs/runtime"),
+                            name="resource.runtime",
+                            title="Runtime Resource",
+                            description="Runtime-scoped resource.",
+                            mimeType="text/plain",
+                        )
+                    ]
+                )
+
+            async def list_prompts(self) -> ListPromptsResult:
+                return ListPromptsResult(
+                    prompts=[
+                        Prompt(
+                            name="prompt.runtime",
+                            title="Runtime Prompt",
+                            description="Runtime-scoped prompt.",
+                            arguments=[],
+                        )
+                    ]
+                )
+
         return _FakeClient()  # type: ignore[return-value]
 
 
@@ -209,6 +240,8 @@ def test_persistent_session_public_runtime_exposes_safe_read_operations() -> Non
     assert "call_tool" in public_operations
     assert "read_resource" in public_operations
     assert "get_prompt" in public_operations
+    assert "list_resources" in public_operations
+    assert "list_prompts" in public_operations
     assert "invoke_method" not in public_operations
     assert "send_notification" not in public_operations
 
@@ -298,3 +331,54 @@ async def test_runtime_pool_reuses_session_for_tool_resource_and_prompt() -> Non
         "prompt.summarize:{'text': 'hello'}"
     )
     assert factory.created_connections == [connection]
+
+
+@pytest.mark.asyncio
+async def test_persistent_session_factory_routes_resource_and_prompt_lists() -> None:
+    factory = _FakeFactory()
+    session = await factory.create(_connection(), None)
+
+    resources = await session.list_resources()
+    prompts = await session.list_prompts()
+    await session.close()
+
+    assert resources[0].name == "resource.runtime"
+    assert resources[0].uri == "fixture://docs/runtime"
+    assert prompts[0].name == "prompt.runtime"
+
+
+@pytest.mark.asyncio
+async def test_runtime_pool_reuses_session_for_resource_and_prompt_lists() -> None:
+    factory = _FakeFactory()
+    pool = McpRuntimePool(factory.create)
+    connection = _connection()
+
+    resources = await pool.list_resources(connection, None)
+    prompts = await pool.list_prompts(connection, None)
+    await pool.close_all()
+
+    assert resources[0].name == "resource.runtime"
+    assert prompts[0].name == "prompt.runtime"
+    assert factory.created_connections == [connection]
+
+
+def test_runtime_pool_satisfies_stateful_protocol_static_shape() -> None:
+    from wf_sources_mcp.sdk import (
+        PromptRuntime,
+        ResourceRuntime,
+        StatefulMcpRuntime,
+        ToolRuntime,
+    )
+
+    factory = _FakeFactory()
+    pool = McpRuntimePool(factory.create)
+
+    tool_runtime: ToolRuntime = pool
+    resource_runtime: ResourceRuntime = pool
+    prompt_runtime: PromptRuntime = pool
+    stateful_runtime: StatefulMcpRuntime = pool
+
+    assert tool_runtime is pool
+    assert resource_runtime is pool
+    assert prompt_runtime is pool
+    assert stateful_runtime is pool

tests/wf_sources_mcp/test_sdk_protocols.py

@@ -82,3 +82,19 @@ def test_stateful_mcp_runtime_protocol_shape() -> None:
 
     assert StatefulMcpRuntime.__name__ == "StatefulMcpRuntime"
     assert ToolExecutor.__name__ == "ToolExecutor"
+
+
+def test_stateful_runtime_protocol_slices_export() -> None:
+    from wf_sources_mcp.sdk import (
+        PromptRuntime,
+        ResourceRuntime,
+        StatefulMcpRuntime,
+        ToolExecutor,
+        ToolRuntime,
+    )
+
+    assert ToolRuntime.__name__ == "ToolRuntime"
+    assert ResourceRuntime.__name__ == "ResourceRuntime"
+    assert PromptRuntime.__name__ == "PromptRuntime"
+    assert ToolExecutor.__name__ == "ToolExecutor"
+    assert StatefulMcpRuntime.__name__ == "StatefulMcpRuntime"