feat: route mcp runtime resource reads
This commit is contained in:
@@ -246,6 +246,10 @@ implementation state.
|
||||
operation queue with request metadata and `McpSourceClient` execution.
|
||||
Public runtime remains tool-call-only; `operation` strings are diagnostics
|
||||
labels, not dispatch.
|
||||
- Completed: persistent MCP runtime can now route `read_resource` through
|
||||
the owner-task queue and `McpSourceClient`. This is intentionally a thin
|
||||
wrapper over the existing source-client facade; prompt/raw method runtime
|
||||
operations remain separate future slices.
|
||||
- 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
|
||||
|
||||
@@ -114,7 +114,10 @@ First slices should move leaf modules only and leave `wf_mcp` re-export shims:
|
||||
10. Complete: persistent MCP runtime owner now routes explicit callables through
|
||||
a generic operation queue with request metadata. The runtime still exposes
|
||||
only `call_tool`; non-tool methods require a separate public-surface slice.
|
||||
11. Upstream transport/discovery/session services.
|
||||
11. Complete: persistent MCP runtime can route `read_resource` through the
|
||||
owner-task queue and `McpSourceClient`. Runtime still does not expose
|
||||
`get_prompt`, raw method invocation, or notifications.
|
||||
12. 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`,
|
||||
|
||||
@@ -42,6 +42,7 @@ class PersistentSessionFactory:
|
||||
connection=connection,
|
||||
auth=auth,
|
||||
call_callback=owner.call_tool,
|
||||
read_resource_callback=owner.read_resource,
|
||||
close_callback=owner.close,
|
||||
)
|
||||
|
||||
@@ -143,6 +144,13 @@ class _SessionOwner:
|
||||
run=lambda client: client.call_tool(tool_name, payload),
|
||||
)
|
||||
|
||||
async def read_resource(self, uri: str) -> dict[str, Any]:
|
||||
"""Submit a resource read through the generic owner-task operation queue."""
|
||||
return await self.submit(
|
||||
operation="read_resource",
|
||||
run=lambda client: client.read_resource(uri),
|
||||
)
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Ask the owner task to close the MCP transport in its own scope."""
|
||||
task = self._task
|
||||
|
||||
@@ -83,6 +83,15 @@ class McpRuntimePool:
|
||||
session = await self.get_session(connection, auth)
|
||||
return await session.call_tool(tool_name, payload)
|
||||
|
||||
async def read_resource(
|
||||
self,
|
||||
connection: McpSourceConnection,
|
||||
auth: AuthRecord | None,
|
||||
uri: str,
|
||||
) -> dict[str, Any]:
|
||||
session = await self.get_session(connection, auth)
|
||||
return await session.read_resource(uri)
|
||||
|
||||
async def close_connection(self, connection_id: str) -> None:
|
||||
current = self._sessions.pop(connection_id, None)
|
||||
if current is not None:
|
||||
|
||||
@@ -5,6 +5,7 @@ from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from mcp.client.session import ClientSession
|
||||
from pydantic import AnyUrl
|
||||
|
||||
from wf_sources_mcp.auth import AuthRecord
|
||||
from wf_sources_mcp.connections import McpSourceConnection
|
||||
@@ -12,6 +13,7 @@ from wf_sources_mcp.sdk import ToolCallResult
|
||||
from wf_sources_mcp.sdk.converters import tool_result_to_call_result
|
||||
|
||||
RawToolCaller = Callable[[str, dict[str, Any]], Awaitable[ToolCallResult]]
|
||||
RawResourceReader = Callable[[str], Awaitable[dict[str, Any]]]
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
@@ -32,6 +34,7 @@ class PersistentMcpSession:
|
||||
auth: AuthRecord | None
|
||||
client: ClientSession | None = None
|
||||
call_callback: RawToolCaller | None = None
|
||||
read_resource_callback: RawResourceReader | None = None
|
||||
close_callback: Callable[[], Awaitable[None]] | None = None
|
||||
|
||||
async def call_tool(
|
||||
@@ -44,6 +47,15 @@ class PersistentMcpSession:
|
||||
return tool_result_to_call_result(result)
|
||||
raise RuntimeError("persistent MCP session has no tool call transport")
|
||||
|
||||
async def read_resource(self, uri: str) -> dict[str, Any]:
|
||||
"""Read an MCP resource through the owner task or injected session."""
|
||||
if self.read_resource_callback is not None:
|
||||
return await self.read_resource_callback(uri)
|
||||
if self.client is not None:
|
||||
result = await self.client.read_resource(AnyUrl(uri))
|
||||
return result.model_dump(by_alias=True, mode="json", exclude_none=True)
|
||||
raise RuntimeError("persistent MCP session has no resource read transport")
|
||||
|
||||
async def close(self) -> None:
|
||||
"""Close the transport/session stack owned by the runtime factory."""
|
||||
if self.close_callback is not None:
|
||||
|
||||
@@ -7,6 +7,7 @@ import pytest
|
||||
from mcp.client.session import ClientSession
|
||||
from mcp.types import CallToolResult as RawCallToolResult
|
||||
from mcp.types import TextContent
|
||||
from pydantic import AnyUrl
|
||||
|
||||
from wf_sources_mcp.auth import AuthRecord
|
||||
from wf_sources_mcp.connections import McpSourceConnection
|
||||
@@ -56,6 +57,14 @@ async def test_persistent_session_raises_without_transport() -> None:
|
||||
await session.call_tool("echo", {})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_persistent_session_raises_without_resource_transport() -> None:
|
||||
session = PersistentMcpSession(connection=_connection(), auth=None)
|
||||
|
||||
with pytest.raises(RuntimeError, match="no resource read transport"):
|
||||
await session.read_resource("test://x")
|
||||
|
||||
|
||||
class _FakeFactory(PersistentSessionFactory):
|
||||
def __init__(self) -> None:
|
||||
self.calls: list[tuple[str, dict[str, object]]] = []
|
||||
@@ -85,6 +94,17 @@ class _FakeFactory(PersistentSessionFactory):
|
||||
) -> RawCallToolResult:
|
||||
return await factory._call_tool(tool_name, payload)
|
||||
|
||||
async def read_resource(self, uri: AnyUrl):
|
||||
return type(
|
||||
"ReadResourceResult",
|
||||
(),
|
||||
{
|
||||
"model_dump": lambda _self, **_kwargs: {
|
||||
"contents": [{"uri": str(uri), "text": "resource text"}]
|
||||
}
|
||||
},
|
||||
)()
|
||||
|
||||
return _FakeClient() # type: ignore[return-value]
|
||||
|
||||
|
||||
@@ -148,7 +168,7 @@ def test_runtime_fingerprint_changes_when_transport_changes() -> None:
|
||||
)
|
||||
|
||||
|
||||
def test_persistent_session_public_runtime_is_tool_call_only() -> None:
|
||||
def test_persistent_session_public_runtime_exposes_only_tool_and_resource_read() -> None:
|
||||
public_operations = {
|
||||
name
|
||||
for name in dir(PersistentMcpSession)
|
||||
@@ -156,7 +176,45 @@ def test_persistent_session_public_runtime_is_tool_call_only() -> None:
|
||||
}
|
||||
|
||||
assert "call_tool" in public_operations
|
||||
assert "read_resource" not in public_operations
|
||||
assert "read_resource" in public_operations
|
||||
assert "get_prompt" not in public_operations
|
||||
assert "invoke_method" not in public_operations
|
||||
assert "send_notification" not in public_operations
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_persistent_session_factory_routes_resource_reads_through_owner() -> None:
|
||||
factory = _FakeFactory()
|
||||
connection = _connection()
|
||||
session = await factory.create(connection, None)
|
||||
|
||||
await session.call_tool("echo", {"text": "one"})
|
||||
resource_payload = await session.read_resource("fixture://docs/welcome")
|
||||
await session.close()
|
||||
|
||||
assert factory.created_connections == [connection]
|
||||
assert factory.calls == [("echo", {"text": "one"})]
|
||||
assert resource_payload == {
|
||||
"contents": [
|
||||
{"uri": "fixture://docs/welcome", "text": "resource text"},
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_runtime_pool_reuses_session_for_tool_and_resource_read() -> None:
|
||||
factory = _FakeFactory()
|
||||
pool = McpRuntimePool(factory.create)
|
||||
connection = _connection()
|
||||
|
||||
tool_result = await pool.call_tool(connection, None, "echo", {"text": "one"})
|
||||
resource_payload = await pool.read_resource(
|
||||
connection,
|
||||
None,
|
||||
"fixture://docs/welcome",
|
||||
)
|
||||
await pool.close_all()
|
||||
|
||||
assert tool_result.output == {"echoed": "one"}
|
||||
assert resource_payload["contents"][0]["text"] == "resource text"
|
||||
assert factory.created_connections == [connection]
|
||||
|
||||
Reference in New Issue
Block a user