refactor: add mcp source client facade

This commit is contained in:
lda
2026-06-07 17:34:39 +07:00 Verified
parent f6fa12a011
commit c3de026d46
9 changed files with 386 additions and 65 deletions
+4
View File
@@ -238,6 +238,10 @@ implementation state.
`McpSdkAdapter` is now canonical in `wf_sources_mcp`; `wf_mcp.sdk.*` `McpSdkAdapter` is now canonical in `wf_sources_mcp`; `wf_mcp.sdk.*`
remains a compatibility shim for old imports. Persistent runtime is still remains a compatibility shim for old imports. Persistent runtime is still
tool-call-only. tool-call-only.
- Completed: shared `McpSourceClient` facade introduced in
`wf_sources_mcp.client`. The one-shot SDK adapter delegates MCP operation
calls and conversion through the facade; persistent runtime remains
tool-call-only until a separate owner-task routing slice.
- 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
@@ -107,7 +107,11 @@ First slices should move leaf modules only and leave `wf_mcp` re-export shims:
compatibility shim. This does not expand persistent runtime; the next compatibility shim. This does not expand persistent runtime; the next
design slice should unify one-shot and persistent client operation handling design slice should unify one-shot and persistent client operation handling
behind a shared source-client facade. behind a shared source-client facade.
9. Upstream transport/discovery/session services. 9. Complete: shared `McpSourceClient` facade introduced in
`wf_sources_mcp.client`. `McpSdkAdapter` now delegates operation handling to
this facade. Persistent runtime still exposes only `call_tool`; expanding it
requires a separate owner-task request routing slice.
10. 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`,
+2 -1
View File
@@ -1,5 +1,6 @@
from __future__ import annotations from __future__ import annotations
from .source_client import McpClientSession, McpSourceClient
from .transport import open_mcp_session from .transport import open_mcp_session
__all__ = ["open_mcp_session"] __all__ = ["McpClientSession", "McpSourceClient", "open_mcp_session"]
+143
View File
@@ -0,0 +1,143 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Protocol
from mcp import ClientResult
from mcp.types import (
CallToolResult,
ClientNotification,
ClientRequest,
ListPromptsResult,
ListResourcesResult,
ListToolsResult,
)
from pydantic import AnyUrl
from wf_sources_mcp.connections import McpSourceConnection
if TYPE_CHECKING:
from wf_sources_mcp.catalog import (
DiscoveredPrompt,
DiscoveredResource,
DiscoveredTool,
)
from wf_sources_mcp.sdk.protocols import ToolCallResult
class McpClientSession(Protocol):
"""Subset of MCP SDK ClientSession operations used by source clients."""
async def list_tools(self) -> ListToolsResult: ...
async def list_resources(self) -> ListResourcesResult: ...
async def list_prompts(self) -> ListPromptsResult: ...
async def read_resource(self, uri: AnyUrl) -> Any: ...
async def get_prompt(
self,
name: str,
arguments: dict[str, str] | None = None,
/,
) -> Any: ...
async def send_request(
self,
request: ClientRequest,
result_type: type[ClientResult],
) -> Any: ...
async def send_notification(self, notification: ClientNotification) -> None: ...
async def call_tool(
self,
name: str,
arguments: dict[str, Any],
/,
) -> CallToolResult: ...
@dataclass(slots=True)
class McpSourceClient:
"""Operation facade over an initialized MCP SDK ClientSession.
This class owns SDK operation calls and conversion to wf_sources_mcp DTOs.
It does not own transport lifetime. One-shot callers enter it through
`open_mcp_session`; persistent runtime owners may wrap the session inside
their owner task in a later slice.
"""
session: McpClientSession
connection: McpSourceConnection
async def list_tools(self) -> list[DiscoveredTool]:
from wf_sources_mcp.sdk.converters import tool_to_discovered
result: ListToolsResult = await self.session.list_tools()
return [tool_to_discovered(tool) for tool in result.tools]
async def list_resources(self) -> list[DiscoveredResource]:
from wf_sources_mcp.sdk.converters import resource_to_discovered
result: ListResourcesResult = await self.session.list_resources()
return [resource_to_discovered(resource) for resource in result.resources]
async def list_prompts(self) -> list[DiscoveredPrompt]:
from wf_sources_mcp.sdk.converters import prompt_to_discovered
result: ListPromptsResult = await self.session.list_prompts()
return [prompt_to_discovered(prompt) for prompt in result.prompts]
async def get_connection_metadata(self) -> dict[str, Any]:
transport = self.connection.transport
return {
"server": self.connection.provider,
"transport": transport.kind if transport is not None else None,
}
async def read_resource(self, uri: str) -> dict[str, Any]:
result = await self.session.read_resource(AnyUrl(uri))
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def get_prompt(
self,
prompt_name: str,
arguments: dict[str, str] | None = None,
) -> dict[str, Any]:
result = await self.session.get_prompt(prompt_name, arguments)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def invoke_method(
self,
method: str,
params: dict[str, Any] | None = None,
) -> dict[str, Any]:
result = await self.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,
method: str,
params: dict[str, Any] | None = None,
) -> None:
await self.session.send_notification(
ClientNotification.model_validate({"method": method, "params": params})
)
async def call_tool(
self,
tool_name: str,
payload: dict[str, Any],
) -> ToolCallResult:
from wf_sources_mcp.sdk.converters import tool_result_to_call_result
result = await self.session.call_tool(tool_name, payload)
return tool_result_to_call_result(result)
__all__ = ["McpClientSession", "McpSourceClient"]
+4
View File
@@ -23,6 +23,10 @@ class PersistentMcpSession:
inside an AnyIO cancel scope and must be called and closed by that same 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 owner task. `client` remains available for simple injected/fake sessions in
tests. `call_tool()` always normalizes SDK results for workflow nodes. 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.
""" """
connection: McpSourceConnection connection: McpSourceConnection
+26 -56
View File
@@ -1,86 +1,65 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import Any 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.auth import AuthRecord
from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource, DiscoveredTool from wf_sources_mcp.catalog import DiscoveredPrompt, DiscoveredResource, DiscoveredTool
from wf_sources_mcp.client import open_mcp_session from wf_sources_mcp.client import McpSourceClient, open_mcp_session
from wf_sources_mcp.connections import McpSourceConnection 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 from .protocols import BackendAdapter, ToolCallResult
class McpSdkAdapter(BackendAdapter): class McpSdkAdapter(BackendAdapter):
"""One-shot MCP client adapter for upstream MCP source operations. """One-shot MCP client adapter for upstream MCP source operations.
This adapter intentionally opens a fresh SDK session per operation. Stateful This adapter intentionally opens a fresh SDK session per operation. It
workflow tool execution is handled by `wf_sources_mcp.runtime`; discovery delegates all MCP operation/conversion details to `McpSourceClient` so the
and admin operations use this simpler one-shot path. same operation facade can later be used inside persistent runtime owners.
""" """
@asynccontextmanager @asynccontextmanager
async def _session( async def _client(
self, self,
connection: McpSourceConnection, connection: McpSourceConnection,
auth: AuthRecord | None, auth: AuthRecord | None,
): ) -> AsyncIterator[McpSourceClient]:
async with open_mcp_session(connection, auth) as session: async with open_mcp_session(connection, auth) as session:
yield session yield McpSourceClient(session=session, connection=connection)
async def list_tools( async def list_tools(
self, self,
connection: McpSourceConnection, connection: McpSourceConnection,
auth: AuthRecord | None, auth: AuthRecord | None,
) -> list[DiscoveredTool]: ) -> list[DiscoveredTool]:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
result: ListToolsResult = await session.list_tools() return await client.list_tools()
return [tool_to_discovered(tool) for tool in result.tools]
async def list_resources( async def list_resources(
self, self,
connection: McpSourceConnection, connection: McpSourceConnection,
auth: AuthRecord | None, auth: AuthRecord | None,
) -> list[DiscoveredResource]: ) -> list[DiscoveredResource]:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
result: ListResourcesResult = await session.list_resources() return await client.list_resources()
return [resource_to_discovered(resource) for resource in result.resources]
async def list_prompts( async def list_prompts(
self, self,
connection: McpSourceConnection, connection: McpSourceConnection,
auth: AuthRecord | None, auth: AuthRecord | None,
) -> list[DiscoveredPrompt]: ) -> list[DiscoveredPrompt]:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
result: ListPromptsResult = await session.list_prompts() return await client.list_prompts()
return [prompt_to_discovered(prompt) for prompt in result.prompts]
async def get_connection_metadata( async def get_connection_metadata(
self, self,
connection: McpSourceConnection, connection: McpSourceConnection,
auth: AuthRecord | None, auth: AuthRecord | None,
) -> dict[str, Any]: ) -> dict[str, Any]:
transport = connection.transport async with self._client(connection, auth) as client:
return { return await client.get_connection_metadata()
"server": connection.provider,
"transport": transport.kind if transport is not None else None,
}
async def read_resource( async def read_resource(
self, self,
@@ -88,9 +67,8 @@ class McpSdkAdapter(BackendAdapter):
auth: AuthRecord | None, auth: AuthRecord | None,
uri: str, uri: str,
) -> dict[str, Any]: ) -> dict[str, Any]:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
result = await session.read_resource(AnyUrl(uri)) return await client.read_resource(uri)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def get_prompt( async def get_prompt(
self, self,
@@ -99,9 +77,8 @@ class McpSdkAdapter(BackendAdapter):
prompt_name: str, prompt_name: str,
arguments: dict[str, str] | None = None, arguments: dict[str, str] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
result = await session.get_prompt(prompt_name, arguments) return await client.get_prompt(prompt_name, arguments)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def invoke_method( async def invoke_method(
self, self,
@@ -110,12 +87,8 @@ class McpSdkAdapter(BackendAdapter):
method: str, method: str,
params: dict[str, Any] | None = None, params: dict[str, Any] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
result = await session.send_request( return await client.invoke_method(method, params)
ClientRequest.model_validate({"method": method, "params": params}),
ClientResult,
)
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
async def send_notification( async def send_notification(
self, self,
@@ -124,10 +97,8 @@ class McpSdkAdapter(BackendAdapter):
method: str, method: str,
params: dict[str, Any] | None = None, params: dict[str, Any] | None = None,
) -> None: ) -> None:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
await session.send_notification( await client.send_notification(method, params)
ClientNotification.model_validate({"method": method, "params": params})
)
async def call_tool( async def call_tool(
self, self,
@@ -136,9 +107,8 @@ class McpSdkAdapter(BackendAdapter):
tool_name: str, tool_name: str,
payload: dict[str, Any], payload: dict[str, Any],
) -> ToolCallResult: ) -> ToolCallResult:
async with self._session(connection, auth) as session: async with self._client(connection, auth) as client:
result = await session.call_tool(tool_name, payload) return await client.call_tool(tool_name, payload)
return tool_result_to_call_result(result)
__all__ = ["McpSdkAdapter"] __all__ = ["McpSdkAdapter"]
+10 -7
View File
@@ -18,6 +18,7 @@ from mcp.types import (
) )
from pydantic import AnyUrl from pydantic import AnyUrl
from wf_sources_mcp.client import McpSourceClient
from wf_sources_mcp.connections import McpSourceConnection from wf_sources_mcp.connections import McpSourceConnection
from wf_sources_mcp.sdk import BackendAdapter, McpSdkAdapter from wf_sources_mcp.sdk import BackendAdapter, McpSdkAdapter
from wf_sources_mcp.transports import StdioSourceTransport from wf_sources_mcp.transports import StdioSourceTransport
@@ -135,12 +136,12 @@ class _FakeSession:
) )
class _SessionContext: class _ClientContext:
def __init__(self, session: _FakeSession) -> None: def __init__(self, client: McpSourceClient) -> None:
self.session = session self.client = client
async def __aenter__(self) -> _FakeSession: async def __aenter__(self) -> McpSourceClient:
return self.session return self.client
async def __aexit__( async def __aexit__(
self, self,
@@ -155,10 +156,12 @@ class _FakeAdapter(McpSdkAdapter):
def __init__(self, session: _FakeSession) -> None: def __init__(self, session: _FakeSession) -> None:
self.fake_session = session self.fake_session = session
def _session(self, connection: McpSourceConnection, auth: object | None): def _client(self, connection: McpSourceConnection, auth: object | None):
assert connection.id == "demo.personal" assert connection.id == "demo.personal"
assert auth is None assert auth is None
return _SessionContext(self.fake_session) return _ClientContext(
McpSourceClient(session=self.fake_session, connection=connection)
)
def test_mcp_sdk_adapter_implements_backend_protocol() -> None: def test_mcp_sdk_adapter_implements_backend_protocol() -> None:
+192
View File
@@ -0,0 +1,192 @@
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.client import McpSourceClient
from wf_sources_mcp.connections import McpSourceConnection
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.requests: list[ClientRequest] = []
self.notifications: list[ClientNotification] = []
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},
)
@pytest.mark.asyncio
async def test_source_client_lists_catalog_items_and_metadata() -> None:
source_client = McpSourceClient(session=_FakeSession(), connection=_connection())
tools = await source_client.list_tools()
resources = await source_client.list_resources()
prompts = await source_client.list_prompts()
metadata = await source_client.get_connection_metadata()
assert tools[0].name == "echo"
assert resources[0].uri == "fixture://docs/welcome"
assert prompts[0].name == "prompt.summarize"
assert metadata == {"server": "demo", "transport": "stdio"}
@pytest.mark.asyncio
async def test_source_client_reads_resources_and_prompts_as_payloads() -> None:
source_client = McpSourceClient(session=_FakeSession(), connection=_connection())
resource_payload = await source_client.read_resource("fixture://docs/welcome")
prompt_payload = await source_client.get_prompt(
"prompt.summarize",
{"text": "hello"},
)
assert resource_payload["contents"][0]["text"] == "hello"
assert prompt_payload["messages"][0]["content"]["text"] == (
"prompt.summarize:{'text': 'hello'}"
)
@pytest.mark.asyncio
async def test_source_client_invokes_methods_and_notifications() -> None:
session = _FakeSession()
source_client = McpSourceClient(session=session, connection=_connection())
result = await source_client.invoke_method("ping")
await source_client.send_notification("notifications/initialized")
assert result == {"ok": True}
assert session.requests, "invoke_method should send a request"
assert session.notifications, "send_notification should send a notification"
@pytest.mark.asyncio
async def test_source_client_call_tool_normalizes_result() -> None:
source_client = McpSourceClient(session=_FakeSession(), connection=_connection())
result = await source_client.call_tool("echo", {"text": "hello"})
assert result.outcome == "ok"
assert result.output == {
"tool": "echo",
"payload": {"text": "hello"},
}