feat: prefer stateful runtime for mcp content access

This commit is contained in:
lda
2026-06-07 19:49:07 +07:00 Verified
parent 76ac0432a0
commit 8bb2f32d9a
14 changed files with 252 additions and 14 deletions
+3
View File
@@ -253,6 +253,9 @@ implementation state.
- Completed: persistent MCP runtime can now route `get_prompt` through - Completed: persistent MCP runtime can now route `get_prompt` through
the owner-task queue and `McpSourceClient`. This keeps prompt reads the owner-task queue and `McpSourceClient`. This keeps prompt reads
stateful without adding raw method invocation or notification support. stateful without adding raw method invocation or notification support.
- 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.
- Auth/source secrets boundary: keep registry desired state separate from - Auth/source secrets boundary: keep registry desired state separate from
upstream credentials, and surface missing auth as validation diagnostics. upstream credentials, and surface missing auth as validation diagnostics.
The contract is now specified in The contract is now specified in
@@ -1,5 +1,10 @@
# Content Access Stateful Runtime Routing Implementation Plan # Content Access Stateful Runtime Routing Implementation Plan
> **Historical:** This plan has been implemented. The `StatefulMcpRuntime` protocol
> is in `wf_sources_mcp.sdk`, upstream transport prefers it for content reads,
> and `WfMcpService` wires the configured runtime pool as both `tool_executor`
> and `stateful_runtime`.
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Route broker content reads (`read_resource`, `render_prompt`) through a stateful MCP runtime when one is configured, while preserving one-shot adapter fallback. **Goal:** Route broker content reads (`read_resource`, `render_prompt`) through a stateful MCP runtime when one is configured, while preserving one-shot adapter fallback.
@@ -120,7 +120,10 @@ First slices should move leaf modules only and leave `wf_mcp` re-export shims:
12. Complete: persistent MCP runtime can route `get_prompt` through the 12. Complete: persistent MCP runtime can route `get_prompt` through the
owner-task queue and `McpSourceClient`. Runtime still does not expose raw owner-task queue and `McpSourceClient`. Runtime still does not expose raw
method invocation, notifications, or discovery list operations. method invocation, notifications, or discovery list operations.
13. Upstream transport/discovery/session services. 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.
Each slice should add import-direction tests so the new source-provider package 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`, does not depend on `wf_mcp.workflow_surface`, `wf_mcp.admin_surface`,
+6 -1
View File
@@ -168,6 +168,10 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
# focused services receive role-specific stores. # focused services receive role-specific stores.
auth_store = FileAuthStore(store_roots.auth_root) auth_store = FileAuthStore(store_roots.auth_root)
catalog_store = FileCatalogStore(store_roots.catalog_cache_root) catalog_store = FileCatalogStore(store_roots.catalog_cache_root)
# Discovery can use short-lived SDK sessions. Workflow execution needs
# a persistent runtime so stateful MCP servers keep session/page state
# across sequential workflow nodes.
runtime_pool = McpRuntimePool(runtime_factory.create)
service = WfMcpService( service = WfMcpService(
store=FileStore(store_roots.auth_root), store=FileStore(store_roots.auth_root),
auth_store=auth_store, auth_store=auth_store,
@@ -178,7 +182,8 @@ def build_service_from_config(config: BrokerConfig) -> WfMcpService:
# Discovery can use short-lived SDK sessions. Workflow execution needs # Discovery can use short-lived SDK sessions. Workflow execution needs
# a persistent runtime so stateful MCP servers keep session/page state # a persistent runtime so stateful MCP servers keep session/page state
# across sequential workflow nodes. # across sequential workflow nodes.
tool_executor=McpRuntimePool(runtime_factory.create), tool_executor=runtime_pool,
stateful_runtime=runtime_pool,
) )
source_registry_store = FileSourceRegistryStore(store_roots.source_registry_root) source_registry_store = FileSourceRegistryStore(store_roots.source_registry_root)
service.sync_connections_from_config( service.sync_connections_from_config(
+3 -1
View File
@@ -26,7 +26,7 @@ from wf_sources_mcp.catalog import (
CatalogPromptEntry, CatalogPromptEntry,
CatalogResourceEntry, CatalogResourceEntry,
) )
from wf_sources_mcp.sdk import BackendAdapter, ToolExecutor from wf_sources_mcp.sdk import BackendAdapter, StatefulMcpRuntime, ToolExecutor
from wf_sources_mcp.source_registry import SourceRegistryStore from wf_sources_mcp.source_registry import SourceRegistryStore
from wf_sources_mcp.storage import AuthStore, CatalogStore, Store from wf_sources_mcp.storage import AuthStore, CatalogStore, Store
@@ -68,6 +68,7 @@ class WfMcpService:
draft_workspace_store: DraftWorkspaceStore | None = None draft_workspace_store: DraftWorkspaceStore | None = None
run_store: RunStore | None = None run_store: RunStore | None = None
tool_executor: ToolExecutor | None = None tool_executor: ToolExecutor | None = None
stateful_runtime: StatefulMcpRuntime | None = None
events: BrokerEventRecorder = field(init=False) events: BrokerEventRecorder = field(init=False)
connection_service: ConnectionService = field(init=False) connection_service: ConnectionService = field(init=False)
upstream: UpstreamTransportService = field(init=False) upstream: UpstreamTransportService = field(init=False)
@@ -91,6 +92,7 @@ class WfMcpService:
catalog_store=catalog_store, catalog_store=catalog_store,
event_sink=self.events.record_event, event_sink=self.events.record_event,
tool_executor=self.tool_executor, tool_executor=self.tool_executor,
stateful_runtime=self.stateful_runtime,
) )
self.source_catalog = SourceCatalogService( self.source_catalog = SourceCatalogService(
store=catalog_store, store=catalog_store,
@@ -28,10 +28,9 @@ from wf_mcp.shared.errors import error_payload
from wf_sources_mcp.auth import AuthRecord, connection_auth_diagnostic from wf_sources_mcp.auth import AuthRecord, connection_auth_diagnostic
from wf_sources_mcp.catalog.models import CatalogSnapshot from wf_sources_mcp.catalog.models import CatalogSnapshot
from wf_sources_mcp.connections import ( from wf_sources_mcp.connections import (
McpSourceConnection,
mcp_source_connection_from_connection_config, mcp_source_connection_from_connection_config,
) )
from wf_sources_mcp.sdk import BackendAdapter, ToolExecutor from wf_sources_mcp.sdk import BackendAdapter, StatefulMcpRuntime, ToolExecutor
from wf_sources_mcp.storage import AuthStore, CatalogStore from wf_sources_mcp.storage import AuthStore, CatalogStore
from .adapters import require_adapter from .adapters import require_adapter
@@ -53,6 +52,7 @@ class UpstreamTransportService:
event_sink: EventSink event_sink: EventSink
adapters: dict[str, BackendAdapter] = field(default_factory=dict) adapters: dict[str, BackendAdapter] = field(default_factory=dict)
tool_executor: ToolExecutor | None = None tool_executor: ToolExecutor | None = None
stateful_runtime: StatefulMcpRuntime | None = None
def register_adapter(self, server: str, adapter: BackendAdapter) -> None: def register_adapter(self, server: str, adapter: BackendAdapter) -> None:
self.adapters[server] = adapter self.adapters[server] = adapter
@@ -103,7 +103,6 @@ class UpstreamTransportService:
qualified_name: str, qualified_name: str,
uri: str, uri: str,
) -> dict[str, Any]: ) -> dict[str, Any]:
adapter = require_adapter(connection, self.adapters)
auth = self.load_connection_auth(connection) auth = self.load_connection_auth(connection)
# Compatibility boundary: broker callers still pass ConnectionConfig. # Compatibility boundary: broker callers still pass ConnectionConfig.
source_connection = mcp_source_connection_from_connection_config(connection) source_connection = mcp_source_connection_from_connection_config(connection)
@@ -115,6 +114,14 @@ class UpstreamTransportService:
payload={"uri": uri}, payload={"uri": uri},
) )
) )
if self.stateful_runtime is not None:
result = await self.stateful_runtime.read_resource(
source_connection,
auth,
uri,
)
else:
adapter = require_adapter(connection, self.adapters)
result = await adapter.read_resource(source_connection, auth, uri) result = await adapter.read_resource(source_connection, auth, uri)
self.event_sink( self.event_sink(
make_event( make_event(
@@ -133,7 +140,6 @@ class UpstreamTransportService:
local_name: str, local_name: str,
arguments: dict[str, str] | None = None, arguments: dict[str, str] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
adapter = require_adapter(connection, self.adapters)
auth = self.load_connection_auth(connection) auth = self.load_connection_auth(connection)
# Compatibility boundary: broker callers still pass ConnectionConfig. # Compatibility boundary: broker callers still pass ConnectionConfig.
source_connection = mcp_source_connection_from_connection_config(connection) source_connection = mcp_source_connection_from_connection_config(connection)
@@ -145,6 +151,15 @@ class UpstreamTransportService:
payload={"argument_keys": sorted((arguments or {}).keys())}, payload={"argument_keys": sorted((arguments or {}).keys())},
) )
) )
if self.stateful_runtime is not None:
result = await self.stateful_runtime.get_prompt(
source_connection,
auth,
local_name,
arguments,
)
else:
adapter = require_adapter(connection, self.adapters)
result = await adapter.get_prompt(source_connection, auth, local_name, arguments) result = await adapter.get_prompt(source_connection, auth, local_name, arguments)
self.event_sink( self.event_sink(
make_event( make_event(
+7 -2
View File
@@ -1,3 +1,8 @@
from wf_sources_mcp.sdk import BackendAdapter, McpSdkAdapter, ToolCallResult from wf_sources_mcp.sdk import (
BackendAdapter,
McpSdkAdapter,
StatefulMcpRuntime,
ToolCallResult,
)
__all__ = ["BackendAdapter", "McpSdkAdapter", "ToolCallResult"] __all__ = ["BackendAdapter", "McpSdkAdapter", "StatefulMcpRuntime", "ToolCallResult"]
+2 -1
View File
@@ -5,9 +5,10 @@ Canonical implementation lives in `wf_sources_mcp.sdk`.
from __future__ import annotations from __future__ import annotations
from wf_sources_mcp.sdk import BackendAdapter, ToolCallResult from wf_sources_mcp.sdk import BackendAdapter, StatefulMcpRuntime, ToolCallResult
__all__ = [ __all__ = [
"BackendAdapter", "BackendAdapter",
"StatefulMcpRuntime",
"ToolCallResult", "ToolCallResult",
] ]
+2 -1
View File
@@ -8,11 +8,12 @@ from .converters import (
tool_to_discovered, tool_to_discovered,
workflow_output_schema_from_mcp_tool_schema, workflow_output_schema_from_mcp_tool_schema,
) )
from .protocols import BackendAdapter, ToolCallResult, ToolExecutor from .protocols import BackendAdapter, StatefulMcpRuntime, ToolCallResult, ToolExecutor
__all__ = [ __all__ = [
"BackendAdapter", "BackendAdapter",
"McpSdkAdapter", "McpSdkAdapter",
"StatefulMcpRuntime",
"ToolCallResult", "ToolCallResult",
"ToolExecutor", "ToolExecutor",
"prompt_to_discovered", "prompt_to_discovered",
+24
View File
@@ -99,8 +99,32 @@ class ToolExecutor(Protocol):
) -> ToolCallResult: ... ) -> ToolCallResult: ...
class StatefulMcpRuntime(ToolExecutor, Protocol):
"""Stateful execution/read boundary for configured MCP sources.
Implementations keep source session state across calls. Discovery/catalog
refresh may still use one-shot adapters by policy.
"""
async def read_resource(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
uri: str,
) -> dict[str, Any]: ...
async def get_prompt(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
prompt_name: str,
arguments: dict[str, str] | None = None,
) -> dict[str, Any]: ...
__all__ = [ __all__ = [
"BackendAdapter", "BackendAdapter",
"StatefulMcpRuntime",
"ToolCallResult", "ToolCallResult",
"ToolExecutor", "ToolExecutor",
] ]
@@ -156,3 +156,59 @@ async def test_content_access_raises_on_unknown_resource() -> None:
with pytest.raises(KeyError): with pytest.raises(KeyError):
await content_access.read_resource("nonexistent.resource") await content_access.read_resource("nonexistent.resource")
class _StatefulRuntime:
def __init__(self) -> None:
self.resources: list[str] = []
self.prompts: list[str] = []
async def call_tool(self, connection, auth, tool_name, payload):
raise AssertionError("not used")
async def read_resource(self, connection, auth, uri: str):
self.resources.append(uri)
return {"contents": [{"uri": uri, "text": "stateful resource"}]}
async def get_prompt(self, connection, auth, prompt_name, arguments=None):
self.prompts.append(prompt_name)
return {
"messages": [
{
"role": "user",
"content": {"type": "text", "text": "stateful prompt"},
}
]
}
async def test_content_access_uses_stateful_runtime_for_upstream_content() -> None:
runtime = _StatefulRuntime()
service = WfMcpService(
store=FileStore(local_temp_root() / "content_stateful_runtime"),
tool_executor=runtime,
stateful_runtime=runtime,
)
service.register_connection(
ConnectionConfig(
id="demo.personal",
server="demo",
account="personal",
metadata={"transport": "stdio", "command": "fake-mcp-server"},
)
)
service.register_adapter("demo", FakeAdapter())
await service.refresh_connection_catalog("demo.personal")
resource = await service.content_access.read_resource(
"demo.personal.resource.welcome"
)
prompt = await service.content_access.render_prompt(
"demo.personal.prompt.summarize",
arguments={"text": "hello"},
)
assert resource["contents"][0]["text"] == "stateful resource"
assert prompt["messages"][0]["content"]["text"] == "stateful prompt"
assert runtime.resources == ["demo://docs/welcome"]
assert runtime.prompts == ["prompt.summarize"]
@@ -351,3 +351,110 @@ def test_upstream_transport_uses_separate_auth_and_catalog_stores(
assert (tmp_path / "auth" / "auth" / "demo.personal.json").exists() assert (tmp_path / "auth" / "auth" / "demo.personal.json").exists()
assert (tmp_path / "catalog" / "catalog" / "demo.personal.json").exists() assert (tmp_path / "catalog" / "catalog" / "demo.personal.json").exists()
class _StatefulRuntime:
def __init__(self) -> None:
self.resources: list[tuple[str, str]] = []
self.prompts: list[tuple[str, str, dict[str, str] | None]] = []
async def call_tool(self, connection, auth, tool_name, payload):
raise AssertionError("not used by these tests")
async def read_resource(self, connection, auth, uri: str):
self.resources.append((connection.id, uri))
return {"contents": [{"uri": uri, "text": "stateful resource"}]}
async def get_prompt(
self,
connection,
auth,
prompt_name: str,
arguments: dict[str, str] | None = None,
):
self.prompts.append((connection.id, prompt_name, arguments))
return {
"messages": [
{
"role": "user",
"content": {"type": "text", "text": "stateful prompt"},
}
]
}
class _ExplodingContentAdapter(FakeAdapter):
async def read_resource(self, connection, auth, uri):
raise AssertionError("adapter read_resource should not be used")
async def get_prompt(self, connection, auth, prompt_name, arguments=None):
raise AssertionError("adapter get_prompt should not be used")
async def test_upstream_transport_prefers_stateful_runtime_for_resource_reads(
tmp_path: Path,
) -> None:
events: list[McpEvent] = []
runtime = _StatefulRuntime()
transport = UpstreamTransportService(
auth_store=FileStore(tmp_path),
catalog_store=FileStore(tmp_path),
event_sink=events.append,
stateful_runtime=runtime,
)
transport.register_adapter("demo", _ExplodingContentAdapter())
connection = ConnectionConfig(
id="demo.personal",
server="demo",
account="personal",
metadata=_fake_transport_metadata(),
)
result = await transport.read_resource(
connection,
"demo.personal.resource.welcome",
"fixture://docs/welcome",
)
assert result["contents"][0]["text"] == "stateful resource"
assert runtime.resources == [("demo.personal", "fixture://docs/welcome")]
assert [event.kind for event in events] == [
"resource_read_started",
"resource_read_completed",
]
async def test_upstream_transport_prefers_stateful_runtime_for_prompts(
tmp_path: Path,
) -> None:
events: list[McpEvent] = []
runtime = _StatefulRuntime()
transport = UpstreamTransportService(
auth_store=FileStore(tmp_path),
catalog_store=FileStore(tmp_path),
event_sink=events.append,
stateful_runtime=runtime,
)
transport.register_adapter("demo", _ExplodingContentAdapter())
connection = ConnectionConfig(
id="demo.personal",
server="demo",
account="personal",
metadata=_fake_transport_metadata(),
)
result = await transport.render_prompt(
connection,
"demo.personal.prompt.summarize",
"prompt.summarize",
{"text": "hello"},
)
assert result["messages"][0]["content"]["text"] == "stateful prompt"
assert runtime.prompts == [
("demo.personal", "prompt.summarize", {"text": "hello"})
]
assert [event.kind for event in events] == [
"prompt_get_started",
"prompt_get_completed",
]
+5 -1
View File
@@ -106,15 +106,19 @@ def test_wf_mcp_catalog_models_shim_reexports_wf_sources_mcp_catalog_models() ->
def test_wf_mcp_sdk_protocol_shims_reexport_wf_sources_mcp_sdk() -> None: def test_wf_mcp_sdk_protocol_shims_reexport_wf_sources_mcp_sdk() -> None:
from wf_mcp.sdk import BackendAdapter as CompatBackendAdapter from wf_mcp.sdk import BackendAdapter as CompatBackendAdapter
from wf_mcp.sdk import StatefulMcpRuntime as CompatStatefulMcpRuntime
from wf_mcp.sdk import ToolCallResult as CompatToolCallResult from wf_mcp.sdk import ToolCallResult as CompatToolCallResult
from wf_mcp.sdk.base import BackendAdapter as CompatBaseBackendAdapter from wf_mcp.sdk.base import BackendAdapter as CompatBaseBackendAdapter
from wf_mcp.sdk.base import StatefulMcpRuntime as CompatBaseStatefulMcpRuntime
from wf_mcp.sdk.base import ToolCallResult as CompatBaseToolCallResult from wf_mcp.sdk.base import ToolCallResult as CompatBaseToolCallResult
from wf_sources_mcp.sdk import BackendAdapter, ToolCallResult from wf_sources_mcp.sdk import BackendAdapter, StatefulMcpRuntime, ToolCallResult
assert CompatBackendAdapter is BackendAdapter assert CompatBackendAdapter is BackendAdapter
assert CompatToolCallResult is ToolCallResult assert CompatToolCallResult is ToolCallResult
assert CompatStatefulMcpRuntime is StatefulMcpRuntime
assert CompatBaseBackendAdapter is BackendAdapter assert CompatBaseBackendAdapter is BackendAdapter
assert CompatBaseToolCallResult is ToolCallResult assert CompatBaseToolCallResult is ToolCallResult
assert CompatBaseStatefulMcpRuntime is StatefulMcpRuntime
def test_wf_mcp_runtime_protocol_shim_reexports_wf_sources_mcp_tool_executor() -> None: def test_wf_mcp_runtime_protocol_shim_reexports_wf_sources_mcp_tool_executor() -> None:
@@ -75,3 +75,10 @@ async def test_tool_executor_protocol_can_describe_tool_calls() -> None:
assert result.outcome == "ok" assert result.outcome == "ok"
assert result.output == {"echoed": {"message": "hello"}} assert result.output == {"echoed": {"message": "hello"}}
def test_stateful_mcp_runtime_protocol_shape() -> None:
from wf_sources_mcp.sdk import StatefulMcpRuntime, ToolExecutor
assert StatefulMcpRuntime.__name__ == "StatefulMcpRuntime"
assert ToolExecutor.__name__ == "ToolExecutor"