lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: move mcp sdk adapter to wf_sources_mcp

Browse Source
This commit is contained in:
lda
2026-06-07 17:07:39 +07:00 Verified
parent 4a15258b6d
commit 977915fd05
13 changed files with 399 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/current_roadmap.md
+10 -6
View File
@@ -228,12 +228,16 @@ implementation state.
- Completed: shared MCP session opener exists in `wf_sources_mcp.client`.
One-shot adapter (`McpSdkAdapter`) and persistent runtime
(`PersistentSessionFactory`) both use it.
- Completed: persistent MCP runtime moved to `wf_sources_mcp.runtime`.
`PersistentMcpSession`, `PersistentSessionFactory`, `McpRuntimePool`,
and `connection_runtime_fingerprint` are now canonical in
`wf_sources_mcp.runtime`; `wf_mcp.runtime.*` are compatibility shims.
Runtime remains tool-call-only. The completed plan was
[2026-06-07 MCP runtime package move](./historical/superpowers/plans/2026-06-07-mcp-runtime-package-move.md).
- Completed: persistent MCP runtime moved to `wf_sources_mcp.runtime`.
`PersistentMcpSession`, `PersistentSessionFactory`, `McpRuntimePool`,
and `connection_runtime_fingerprint` are now canonical in
`wf_sources_mcp.runtime`; `wf_mcp.runtime.*` are compatibility shims.
Runtime remains tool-call-only. The completed plan was
[2026-06-07 MCP runtime package move](./historical/superpowers/plans/2026-06-07-mcp-runtime-package-move.md).
- Completed: one-shot MCP SDK adapter moved to `wf_sources_mcp.sdk.adapter`.
`McpSdkAdapter` is now canonical in `wf_sources_mcp`; `wf_mcp.sdk.*`
remains a compatibility shim for old imports. Persistent runtime is still
tool-call-only.
- 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/plans/2026-06-07-mcp-sdk-adapter-package-move.md → docs/historical/superpowers/plans/2026-06-07-mcp-sdk-adapter-package-move.md
View File
docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md
+6 -1
View File
@@ -102,7 +102,12 @@ First slices should move leaf modules only and leave `wf_mcp` re-export shims:
`wf_mcp.runtime.*` retained as compatibility shims. Runtime remains
tool-call-only. Next slice is moving `McpSdkAdapter` to
`wf_sources_mcp.sdk.adapter`.
8. Upstream transport/discovery/session services.
8. Complete: one-shot MCP SDK adapter moved to
`wf_sources_mcp.sdk.adapter`, with `wf_mcp.sdk.adapter` retained as a
compatibility shim. This does not expand persistent runtime; the next
design slice should unify one-shot and persistent client operation handling
behind a shared source-client facade.
9. 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/broker/config.py
+1 -1
View File
@@ -7,6 +7,7 @@ from wf_api import file_workflow_stores
from wf_config import WorkflowConfigFile
from wf_config.models import FilesystemStoreConfig, McpSourceConfig, ServerConfig
from wf_sources_mcp.runtime import McpRuntimePool, PersistentSessionFactory
from wf_sources_mcp.sdk import McpSdkAdapter
from wf_sources_mcp.source_registry import (
FileSourceRegistryStore,
workflow_mcp_source_to_connection_config,
@@ -15,7 +16,6 @@ from wf_sources_mcp.storage import FileAuthStore, FileCatalogStore, FileStore
from ..control import BrokerConfigFile, ConnectionConfigFile
from ..models import BrokerConfig
from ..sdk import McpSdkAdapter
from .models import BrokerStoreRoots
from .service import WfMcpService
src/wf_mcp/broker/server.py
+1 -1
View File
@@ -12,10 +12,10 @@ from wf_api import (
from wf_api.stores import WorkflowStores
from wf_config import WorkflowConfigFile
from wf_server import WorkflowServer, WorkflowServerConfig
from wf_sources_mcp.sdk import McpSdkAdapter
from wf_sources_mcp.source_registry import FileSourceRegistryStore, SourceRegistryStore
from ..models import BrokerConfig
from ..sdk.adapter import McpSdkAdapter
from .artifact_tools import register_artifact_tools
from .config import broker_config_from_workflow_config, build_service_from_config
from .prompts import register_broker_prompts
src/wf_mcp/sdk/__init__.py
+1 -3
View File
@@ -1,5 +1,3 @@
from wf_sources_mcp.sdk import BackendAdapter, ToolCallResult
from .adapter import McpSdkAdapter
from wf_sources_mcp.sdk import BackendAdapter, McpSdkAdapter, ToolCallResult
__all__ = ["BackendAdapter", "McpSdkAdapter", "ToolCallResult"]
src/wf_mcp/sdk/adapter.py
+3 -131
View File
@@ -1,133 +1,5 @@
from __future__ import annotations
"""Compatibility shim for the canonical MCP SDK adapter."""
from contextlib import asynccontextmanager
from typing import Any
from wf_sources_mcp.sdk.adapter import McpSdkAdapter
from mcp import ClientResult
from mcp.types import (
ClientNotification,
ClientRequest,
ListPromptsResult,
ListResourcesResult,
ListToolsResult,
)
from pydantic import AnyUrl
from wf_sources_mcp.auth import AuthRecord
from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource, DiscoveredTool
from wf_sources_mcp.client import open_mcp_session
from wf_sources_mcp.connections import McpSourceConnection
from wf_sources_mcp.sdk import BackendAdapter, ToolCallResult
from wf_sources_mcp.sdk.converters import (
prompt_to_discovered,
resource_to_discovered,
tool_result_to_call_result,
tool_to_discovered,
)
class McpSdkAdapter(BackendAdapter):
@asynccontextmanager
async def _session(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
):
async with open_mcp_session(connection, auth) as session:
yield session
async def list_tools(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> list[DiscoveredTool]:
async with self._session(connection, auth) as session:
result: ListToolsResult = await session.list_tools()
return [tool_to_discovered(tool) for tool in result.tools]
async def list_resources(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> list[DiscoveredResource]:
async with self._session(connection, auth) as session:
result: ListResourcesResult = await session.list_resources()
return [resource_to_discovered(resource) for resource in result.resources]
async def list_prompts(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> list[DiscoveredPrompt]:
async with self._session(connection, auth) as session:
result: ListPromptsResult = await session.list_prompts()
return [prompt_to_discovered(prompt) for prompt in result.prompts]
async def get_connection_metadata(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> dict[str, Any]:
transport = connection.transport
return {
"server": connection.provider,
"transport": transport.kind if transport is not None else None,
}
async def read_resource(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
uri: str,
) -> dict[str, Any]:
async with self._session(connection, auth) as session:
result = await session.read_resource(AnyUrl(uri))
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def get_prompt(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
prompt_name: str,
arguments: dict[str, str] | None = None,
) -> dict[str, Any]:
async with self._session(connection, auth) as session:
result = await session.get_prompt(prompt_name, arguments)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def invoke_method(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
method: str,
params: dict[str, Any] | None = None,
) -> dict[str, Any]:
async with self._session(connection, auth) as session:
result = await session.send_request(
ClientRequest.model_validate({"method": method, "params": params}),
ClientResult,
)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def send_notification(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
method: str,
params: dict[str, Any] | None = None,
) -> None:
async with self._session(connection, auth) as session:
await session.send_notification(
ClientNotification.model_validate({"method": method, "params": params})
)
async def call_tool(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
tool_name: str,
payload: dict[str, Any],
) -> ToolCallResult:
async with self._session(connection, auth) as session:
result = await session.call_tool(tool_name, payload)
return tool_result_to_call_result(result)
__all__ = ["McpSdkAdapter"]
src/wf_mcp/server/core.py
+1 -1
View File
@@ -7,6 +7,7 @@ from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.client.transports.memory import FastMCPTransport
from wf_sources_mcp.sdk import McpSdkAdapter
from wf_sources_mcp.source_registry import FileSourceRegistryStore
from ..admin_surface import register_service_admin_tools
@@ -15,7 +16,6 @@ from ..broker.transport import normalize_transport
from ..documentation import build_local_documentation_source
from ..models import BrokerConfig
from ..proxy.runtime import ProxyRuntime
from ..sdk import McpSdkAdapter
from ..workflow_surface import register_workflow_tools
from .prompts import register_documentation_prompts
from .resources import register_documentation_resources
src/wf_sources_mcp/sdk/__init__.py
+2
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
from .adapter import McpSdkAdapter
from .converters import (
prompt_to_discovered,
resource_to_discovered,
@@ -11,6 +12,7 @@ from .protocols import BackendAdapter, ToolCallResult, ToolExecutor
__all__ = [
"BackendAdapter",
"McpSdkAdapter",
"ToolCallResult",
"ToolExecutor",
"prompt_to_discovered",
src/wf_sources_mcp/sdk/adapter.py
+144
View File
@@ -0,0 +1,144 @@
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import Any
from mcp import ClientResult
from mcp.types import (
ClientNotification,
ClientRequest,
ListPromptsResult,
ListResourcesResult,
ListToolsResult,
)
from pydantic import AnyUrl
from wf_sources_mcp.auth import AuthRecord
from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource, DiscoveredTool
from wf_sources_mcp.client import open_mcp_session
from wf_sources_mcp.connections import McpSourceConnection
from .converters import (
prompt_to_discovered,
resource_to_discovered,
tool_result_to_call_result,
tool_to_discovered,
)
from .protocols import BackendAdapter, ToolCallResult
class McpSdkAdapter(BackendAdapter):
"""One-shot MCP client adapter for upstream MCP source operations.
This adapter intentionally opens a fresh SDK session per operation. Stateful
workflow tool execution is handled by `wf_sources_mcp.runtime`; discovery
and admin operations use this simpler one-shot path.
"""
@asynccontextmanager
async def _session(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
):
async with open_mcp_session(connection, auth) as session:
yield session
async def list_tools(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> list[DiscoveredTool]:
async with self._session(connection, auth) as session:
result: ListToolsResult = await session.list_tools()
return [tool_to_discovered(tool) for tool in result.tools]
async def list_resources(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> list[DiscoveredResource]:
async with self._session(connection, auth) as session:
result: ListResourcesResult = await session.list_resources()
return [resource_to_discovered(resource) for resource in result.resources]
async def list_prompts(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> list[DiscoveredPrompt]:
async with self._session(connection, auth) as session:
result: ListPromptsResult = await session.list_prompts()
return [prompt_to_discovered(prompt) for prompt in result.prompts]
async def get_connection_metadata(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
) -> dict[str, Any]:
transport = connection.transport
return {
"server": connection.provider,
"transport": transport.kind if transport is not None else None,
}
async def read_resource(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
uri: str,
) -> dict[str, Any]:
async with self._session(connection, auth) as session:
result = await session.read_resource(AnyUrl(uri))
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def get_prompt(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
prompt_name: str,
arguments: dict[str, str] | None = None,
) -> dict[str, Any]:
async with self._session(connection, auth) as session:
result = await session.get_prompt(prompt_name, arguments)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def invoke_method(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
method: str,
params: dict[str, Any] | None = None,
) -> dict[str, Any]:
async with self._session(connection, auth) as session:
result = await session.send_request(
ClientRequest.model_validate({"method": method, "params": params}),
ClientResult,
)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def send_notification(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
method: str,
params: dict[str, Any] | None = None,
) -> None:
async with self._session(connection, auth) as session:
await session.send_notification(
ClientNotification.model_validate({"method": method, "params": params})
)
async def call_tool(
self,
connection: McpSourceConnection,
auth: AuthRecord | None,
tool_name: str,
payload: dict[str, Any],
) -> ToolCallResult:
async with self._session(connection, auth) as session:
result = await session.call_tool(tool_name, payload)
return tool_result_to_call_result(result)
__all__ = ["McpSdkAdapter"]
tests/wf_mcp/test_compat_imports.py
+11
View File
@@ -149,6 +149,17 @@ def test_wf_mcp_sdk_converter_shim_reexports_wf_sources_mcp_converters() -> None
assert compat_output_schema is workflow_output_schema_from_mcp_tool_schema
def test_wf_mcp_sdk_adapter_shim_reexports_wf_sources_mcp_adapter() -> None:
from wf_mcp.sdk import McpSdkAdapter as CompatPackageAdapter
from wf_mcp.sdk.adapter import McpSdkAdapter as CompatModuleAdapter
from wf_sources_mcp.sdk import McpSdkAdapter
from wf_sources_mcp.sdk.adapter import McpSdkAdapter as CanonicalModuleAdapter
assert CompatPackageAdapter is McpSdkAdapter
assert CompatModuleAdapter is McpSdkAdapter
assert CanonicalModuleAdapter is McpSdkAdapter
def test_runtime_shims_reexport_wf_sources_mcp_runtime() -> None:
from wf_mcp.runtime import (
McpRuntimePool as OldMcpRuntimePool,
tests/wf_sources_mcp/test_import_direction_guard.py
+1
View File
@@ -78,6 +78,7 @@ def test_wf_sources_mcp_does_not_import_old_sdk_protocol_modules() -> None:
root = Path(__file__).resolve().parents[2] / "src" / "wf_sources_mcp"
forbidden = {
"wf_mcp.sdk",
"wf_mcp.sdk.adapter",
"wf_mcp.sdk.base",
"wf_mcp.runtime",
"wf_mcp.runtime.protocols",
tests/wf_sources_mcp/test_sdk_adapter.py
+218
View File
@@ -0,0 +1,218 @@
from __future__ import annotations
from typing import Any
import pytest
from mcp import ClientResult
from mcp.types import (
CallToolResult,
ClientNotification,
ClientRequest,
ListPromptsResult,
ListResourcesResult,
ListToolsResult,
Prompt,
Resource,
TextContent,
Tool,
)
from pydantic import AnyUrl
from wf_sources_mcp.connections import McpSourceConnection
from wf_sources_mcp.sdk import BackendAdapter, McpSdkAdapter
from wf_sources_mcp.transports import StdioSourceTransport
def _connection() -> McpSourceConnection:
return McpSourceConnection(
id="demo.personal",
provider="demo",
account="personal",
transport=StdioSourceTransport(command="fake"),
)
class _FakeSession:
def __init__(self) -> None:
self.notifications: list[ClientNotification] = []
self.requests: list[ClientRequest] = []
async def list_tools(self) -> ListToolsResult:
return ListToolsResult(
tools=[
Tool(
name="echo",
title="Echo",
description="Echo text.",
inputSchema={"type": "object", "properties": {}},
)
]
)
async def list_resources(self) -> ListResourcesResult:
return ListResourcesResult(
resources=[
Resource(
uri=AnyUrl("fixture://docs/welcome"),
name="resource.welcome",
title="Welcome",
description="Welcome resource.",
mimeType="text/plain",
)
]
)
async def list_prompts(self) -> ListPromptsResult:
return ListPromptsResult(
prompts=[
Prompt(
name="prompt.summarize",
title="Summarize",
description="Summarize input.",
arguments=[],
)
]
)
async def read_resource(self, uri: AnyUrl) -> Any:
return type(
"ReadResourceResult",
(),
{
"model_dump": lambda _self, **_kwargs: {
"contents": [{"uri": str(uri), "text": "hello"}]
}
},
)()
async def get_prompt(
self,
prompt_name: str,
arguments: dict[str, str] | None = None,
) -> Any:
return type(
"GetPromptResult",
(),
{
"model_dump": lambda _self, **_kwargs: {
"messages": [
{
"role": "user",
"content": {
"type": "text",
"text": f"{prompt_name}:{arguments or {}}",
},
}
]
}
},
)()
async def send_request(
self,
request: ClientRequest,
result_type: type[ClientResult],
) -> Any:
assert result_type is ClientResult
self.requests.append(request)
return type(
"ClientResultModel",
(),
{"model_dump": lambda _self, **_kwargs: {"ok": True}},
)()
async def send_notification(self, notification: ClientNotification) -> None:
self.notifications.append(notification)
async def call_tool(
self,
tool_name: str,
payload: dict[str, Any],
) -> CallToolResult:
return CallToolResult(
content=[TextContent(type="text", text="ok")],
structuredContent={"tool": tool_name, "payload": payload},
)
class _SessionContext:
def __init__(self, session: _FakeSession) -> None:
self.session = session
async def __aenter__(self) -> _FakeSession:
return self.session
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: object | None,
) -> None:
return None
class _FakeAdapter(McpSdkAdapter):
def __init__(self, session: _FakeSession) -> None:
self.fake_session = session
def _session(self, connection: McpSourceConnection, auth: object | None):
assert connection.id == "demo.personal"
assert auth is None
return _SessionContext(self.fake_session)
def test_mcp_sdk_adapter_implements_backend_protocol() -> None:
adapter: BackendAdapter = McpSdkAdapter()
assert adapter.__class__.__name__ == "McpSdkAdapter"
@pytest.mark.asyncio
async def test_mcp_sdk_adapter_uses_session_for_all_backend_methods() -> None:
session = _FakeSession()
adapter = _FakeAdapter(session)
connection = _connection()
tools = await adapter.list_tools(connection, None)
resources = await adapter.list_resources(connection, None)
prompts = await adapter.list_prompts(connection, None)
metadata = await adapter.get_connection_metadata(connection, None)
resource_payload = await adapter.read_resource(
connection,
None,
"fixture://docs/welcome",
)
prompt_payload = await adapter.get_prompt(
connection,
None,
"prompt.summarize",
{"text": "hello"},
)
tool_result = await adapter.call_tool(connection, None, "echo", {"text": "hello"})
method_payload = await adapter.invoke_method(
connection,
None,
"ping",
)
await adapter.send_notification(
connection,
None,
"notifications/initialized",
)
assert tools[0].name == "echo"
assert resources[0].uri == "fixture://docs/welcome"
assert prompts[0].name == "prompt.summarize"
assert metadata == {"server": "demo", "transport": "stdio"}
assert resource_payload["contents"][0]["text"] == "hello"
assert prompt_payload["messages"][0]["content"]["text"] == (
"prompt.summarize:{'text': 'hello'}"
)
assert method_payload == {"ok": True}
assert session.requests, "invoke_method should have sent a request"
assert session.notifications, "send_notification should have sent a notification"
assert tool_result.outcome == "ok"
assert tool_result.output == {
"tool": "echo",
"payload": {"text": "hello"},
}