fix: preserve mcp source compatibility

2026-06-07 14:07:22 +07:00
parent 195a967527
commit ac2163bf30
8 changed files with 323 additions and 17 deletions
@@ -221,10 +221,14 @@ implementation state.
     MCP-backed JSON-RPC path. A neutral-config `WorkflowServer` can start an
     interrupting run, be rebuilt from the same filesystem stores, inspect the
     interrupted run, and resume it to completion through `RpcWorkflowApiClient`.
-    - Completed: MCP upstream source runtime cleanup now starts with a typed
-      `McpSourceConnection` seam in `wf_sources_mcp`, not by moving
-      `runtime/factory.py` as-is. The active plan was
-      [2026-06-07 MCP source connection seam](./historical/superpowers/plans/2026-06-07-mcp-source-connection-seam.md).
+   - Completed: MCP upstream source runtime cleanup now starts with a typed
+     `McpSourceConnection` seam in `wf_sources_mcp`, not by moving
+     `runtime/factory.py` as-is. The active plan was
+     [2026-06-07 MCP source connection seam](./historical/superpowers/plans/2026-06-07-mcp-source-connection-seam.md).
+   - Planned next: share one MCP session opener between the one-shot SDK
+     adapter and persistent runtime before moving runtime files. The active
+     plan is
+     [2026-06-07 MCP client session opener](./superpowers/plans/2026-06-07-mcp-client-session-opener.md).
  - 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
@@ -0,0 +1,229 @@
+# MCP Client Session Opener Implementation Plan
+
+> **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:** Create the clean source-provider MCP session opener in `wf_sources_mcp` and make one-shot adapter + persistent runtime borrow it, without moving runtime files yet.
+
+**Architecture:** The previous slice introduced `McpSourceConnection`. This slice makes that seam useful by centralizing transport opening, auth injection, and `ClientSession.initialize()` in `wf_sources_mcp.client`. Existing `wf_mcp.sdk.adapter.McpSdkAdapter` and `wf_mcp.runtime.factory.PersistentSessionFactory` should call the shared opener instead of duplicating stdio/http setup.
+
+**Tech Stack:** Python 3.14, MCP Python SDK, httpx, AnyIO/MCP async context managers, pytest, ruff, basedpyright.
+
+---
+
+## Design Intent
+
+Do not make a prettier copy of `src/wf_mcp/runtime/factory.py`. Keep the good ideas and isolate them:
+
+- keep the actor/owner-task pattern for persistent sessions
+- keep runtime fingerprinting behavior
+- centralize transport opening and auth
+- keep `wf_mcp.runtime.*` in place for now
+- make future movement to `wf_sources_mcp.runtime` mechanical
+
+The desired flow:
+
+```text
+ConnectionConfig
+  -> mcp_source_connection_from_connection_config()
+  -> open_mcp_session(McpSourceConnection, AuthRecord | None)
+  -> ClientSession
+
+McpSdkAdapter: opens per operation
+PersistentSessionFactory: opens once inside owner task
+```
+
+---
+
+## Important Constraint
+
+`StdioSourceTransport.cwd` exists because old persistent runtime supports `metadata["cwd"]`. The shared opener must preserve that field. Otherwise stdio servers that need a working directory regress.
+
+---
+
+## Task 1: Lock Down `cwd` Propagation
+
+**Files:**
+- Modify: `src/wf_sources_mcp/transports.py`
+- Modify: `src/wf_sources_mcp/connections.py`
+- Test: `tests/wf_sources_mcp/test_connections.py`
+
+- [ ] Confirm `StdioSourceTransport` exposes `cwd: str | None = None`.
+- [ ] Confirm legacy `ConnectionConfig` conversion carries `metadata["cwd"]` into `StdioSourceTransport.cwd`.
+- [ ] Confirm tests assert cwd round-trips from legacy connection metadata.
+- [ ] Run:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_connections.py tests/wf_sources_mcp/test_source_registry.py -q
+uv run basedpyright --level error src/wf_sources_mcp
+```
+
+Expected: pass.
+
+---
+
+## Task 2: Create `wf_sources_mcp.client.transport`
+
+**Files:**
+- Create: `src/wf_sources_mcp/client/__init__.py`
+- Create: `src/wf_sources_mcp/client/transport.py`
+- Test: `tests/wf_sources_mcp/test_client_transport.py`
+
+Implement:
+
+```python
+@asynccontextmanager
+async def open_mcp_session(
+    connection: McpSourceConnection,
+    auth: AuthRecord | None,
+) -> AsyncIterator[ClientSession]:
+    ...
+```
+
+Behavior:
+
+- for `StdioSourceTransport`:
+  - merge `transport.env` with `mcp_auth_env(auth)`, auth wins on duplicate keys
+  - pass `command`, `args`, `env`, and `cwd` to `StdioServerParameters`
+  - enter `stdio_client`
+  - enter `ClientSession`
+  - call `await session.initialize()`
+  - yield initialized session
+
+- for `HttpSourceTransport`:
+  - create `httpx.AsyncClient(headers=mcp_auth_headers(auth) or None)`
+  - enter `streamable_http_client(str(transport.url), http_client=http_client)`
+  - enter `ClientSession`
+  - call `await session.initialize()`
+  - yield initialized session
+
+- unsupported transport:
+  - raise `ValueError(f"unsupported MCP transport {transport.kind!r}")`
+
+Testing guidance:
+
+- Use monkeypatch/fakes for `stdio_client`, `streamable_http_client`, and `ClientSession`.
+- Do not start real subprocesses.
+- Assert stdio env merge and cwd propagation.
+- Assert HTTP headers propagation.
+- Assert initialize is called before yielding.
+
+Verification:
+
+```bash
+uv run pytest tests/wf_sources_mcp/test_client_transport.py -q
+uv run ruff check src/wf_sources_mcp tests/wf_sources_mcp
+uv run basedpyright --level error src/wf_sources_mcp
+```
+
+---
+
+## Task 3: Make `McpSdkAdapter` Use The Shared Opener
+
+**Files:**
+- Modify: `src/wf_mcp/sdk/adapter.py`
+- Test: `tests/wf_mcp/test_sdk_adapter.py`
+
+Replace the private `_session()` transport-opening logic with:
+
+```python
+from wf_sources_mcp.client import open_mcp_session
+```
+
+Then each operation should do:
+
+```python
+async with open_mcp_session(connection, auth) as session:
+    ...
+```
+
+Do not move `McpSdkAdapter` yet. This slice only removes duplicated opening logic.
+
+Verification:
+
+```bash
+uv run pytest tests/wf_mcp/test_sdk_adapter.py tests/wf_sources_mcp/test_client_transport.py -q
+uv run basedpyright --level error src/wf_mcp/sdk src/wf_sources_mcp
+```
+
+---
+
+## Task 4: Make `PersistentSessionFactory` Use The Shared Opener
+
+**Files:**
+- Modify: `src/wf_mcp/runtime/factory.py`
+- Test: `tests/wf_mcp/test_stateful_runtime.py`
+
+The factory still receives legacy `ConnectionConfig`. Convert inside `_create_with_stack()`:
+
+```python
+source_connection = mcp_source_connection_from_connection_config(connection)
+```
+
+Then use the shared opener while preserving `AsyncExitStack` ownership:
+
+```python
+session = await stack.enter_async_context(open_mcp_session(source_connection, auth))
+return session
+```
+
+Do not remove `_SessionOwner`. The owner-task pattern is the important fix for AnyIO/MCP cancel-scope ownership.
+
+Verification:
+
+```bash
+uv run pytest tests/wf_mcp/test_stateful_runtime.py -q
+uv run basedpyright --level error src/wf_mcp/runtime src/wf_sources_mcp
+```
+
+---
+
+## Task 5: Verify Boundary And Update Docs
+
+**Files:**
+- Modify: `docs/current_roadmap.md`
+- Modify: `docs/superpowers/specs/2026-06-03-long-lived-workflow-api-boundary.md`
+
+Docs should say:
+
+- shared MCP session opener exists in `wf_sources_mcp.client`
+- one-shot adapter and persistent runtime both use it
+- runtime files are still in `wf_mcp` for compatibility
+- next slice can move `PersistentSessionFactory`, `PersistentMcpSession`, and `McpRuntimePool` to `wf_sources_mcp.runtime`
+
+Final verification:
+
+```bash
+uv run pytest tests/wf_sources_mcp tests/wf_mcp/test_sdk_adapter.py tests/wf_mcp/test_stateful_runtime.py tests/wf_mcp/service/test_upstream_transport.py -q
+uv run ruff check src tests
+uv run basedpyright --level error src
+git diff --check
+```
+
+Expected:
+
+- focused tests pass
+- ruff passes
+- basedpyright has 0 errors
+- no whitespace errors
+
+---
+
+## Non-Goals
+
+- Do not move runtime files yet.
+- Do not move `McpSdkAdapter` yet.
+- Do not introduce WebSocket/SSE support.
+- Do not implement reconnect/backoff policy.
+- Do not broaden persistent runtime beyond existing `call_tool` behavior in this slice.
+- Do not change proxy/frontend MCP code.
+
+---
+
+## Future Slice
+
+After this plan:
+
+1. Move `PersistentSessionFactory`, `PersistentMcpSession`, and `McpRuntimePool` into `wf_sources_mcp.runtime`.
+2. Keep `wf_mcp.runtime.*` shims.
+3. Then move `McpSdkAdapter` into `wf_sources_mcp.sdk.adapter`.
+4. Only after those moves, consider a broader `McpClientSession` abstraction for persistent `read_resource`, `get_prompt`, `invoke_method`, and `send_notification`.
@@ -6,12 +6,15 @@ from dataclasses import asdict, dataclass, field
 from inspect import isawaitable
 from typing import Any, cast

+from wf_sources_mcp.connections import McpSourceConnection
 from wf_sources_mcp.sdk import ToolCallResult
+from wf_sources_mcp.transports import HttpSourceTransport, StdioSourceTransport

 from ..auth import AuthRecord
 from ..models import ConnectionConfig
 from .session import PersistentMcpSession

+RuntimeConnection = ConnectionConfig | McpSourceConnection
 SessionFactory = Callable[
    [ConnectionConfig, AuthRecord | None],
    PersistentMcpSession | Awaitable[PersistentMcpSession],
@@ -19,7 +22,7 @@ SessionFactory = Callable[


 def connection_runtime_fingerprint(
-    connection: ConnectionConfig,
+    connection: RuntimeConnection,
    auth: AuthRecord | None = None,
 ) -> str:
    """Return the connection identity that decides MCP runtime reuse.
@@ -55,7 +58,7 @@ class McpRuntimePool:

    async def get_session(
        self,
-        connection: ConnectionConfig,
+        connection: RuntimeConnection,
        auth: AuthRecord | None,
    ) -> PersistentMcpSession:
        fingerprint = connection_runtime_fingerprint(connection, auth)
@@ -65,7 +68,10 @@ class McpRuntimePool:
        if current is not None:
            await current[1].close()

-        created = self.session_factory(connection, auth)
+        # Compatibility boundary: wrappers now pass McpSourceConnection, while
+        # PersistentSessionFactory still consumes the legacy broker DTO until
+        # the shared opener/runtime move lands.
+        created = self.session_factory(_legacy_connection_config(connection), auth)
        if isawaitable(created):
            session = await created
        else:
@@ -75,8 +81,8 @@ class McpRuntimePool:

    async def call_tool(
        self,
-        connection,
-        auth,
+        connection: RuntimeConnection,
+        auth: AuthRecord | None,
        tool_name: str,
        payload: dict[str, Any],
    ) -> ToolCallResult:
@@ -94,3 +100,45 @@ class McpRuntimePool:
        self._sessions.clear()
        for _fingerprint, session in sessions:
            await session.close()
+
+
+def _legacy_connection_config(connection: RuntimeConnection) -> ConnectionConfig:
+    if isinstance(connection, ConnectionConfig):
+        return connection
+
+    metadata = dict(connection.metadata)
+    transport = connection.transport
+    if isinstance(transport, StdioSourceTransport):
+        metadata.update(
+            {
+                "transport": "stdio",
+                "command": transport.command,
+                "args": list(transport.args),
+                "env": dict(transport.env),
+            }
+        )
+        if transport.cwd is not None:
+            metadata["cwd"] = transport.cwd
+    elif isinstance(transport, HttpSourceTransport):
+        metadata.update(
+            {
+                "transport": "streamable_http",
+                "url": str(transport.url),
+                "headers": dict(transport.headers),
+            }
+        )
+    else:
+        raise ValueError(f"connection {connection.id!r} requires metadata.transport")
+
+    if connection.profile is not None:
+        metadata["profile"] = connection.profile
+    if connection.auth_ref is not None:
+        metadata["auth_ref"] = connection.auth_ref
+
+    return ConnectionConfig(
+        id=connection.id,
+        server=connection.provider,
+        account=connection.account,
+        enabled=connection.enabled,
+        metadata=metadata,
+    )
@@ -38,6 +38,8 @@ class McpSdkAdapter(BackendAdapter):
        auth: AuthRecord | None,
    ):
        transport = connection.transport
+        if transport is None:
+            raise ValueError(f"connection {connection.id!r} requires metadata.transport")
        if isinstance(transport, StdioSourceTransport):
            auth_env = mcp_auth_env(auth)
            env = dict(transport.env)
@@ -67,7 +69,7 @@ class McpSdkAdapter(BackendAdapter):
                        yield session
            return

-        raise ValueError(f"unsupported MCP transport {transport.kind!r}")
+        raise ValueError(f"unsupported MCP transport {type(transport).__name__}")

    async def list_tools(
        self,
@@ -101,9 +103,10 @@ class McpSdkAdapter(BackendAdapter):
        connection: McpSourceConnection,
        auth: AuthRecord | None,
    ) -> dict[str, Any]:
+        transport = connection.transport
        return {
            "server": connection.provider,
-            "transport": connection.transport.kind,
+            "transport": transport.kind if transport is not None else None,
        }

    async def read_resource(
@@ -39,7 +39,7 @@ class McpSourceConnection:
    id: str
    provider: str
    account: str
-    transport: SourceTransport
+    transport: SourceTransport | None = None
    enabled: bool = True
    profile: str | None = None
    auth_ref: str | None = None
@@ -56,6 +56,16 @@ class McpSourceConnection:
                "MCP source connection id must match provider/account fields"
            )

+    @property
+    def server(self) -> str:
+        """Compatibility alias for older adapter code.
+
+        `provider` is the source-provider term. The old broker DTO called the
+        same field `server`, and several fake/custom adapters still read it.
+        """
+
+        return self.provider
+

 def mcp_source_connection_from_registry_entry(
    entry: McpSourceRegistryEntry,
@@ -103,7 +113,9 @@ def mcp_source_connection_from_connection_config(
    )


-def _transport_from_connection_metadata(connection: ConnectionConfig) -> SourceTransport:
+def _transport_from_connection_metadata(
+    connection: ConnectionConfig,
+) -> SourceTransport | None:
    transport = connection.metadata.get("transport")
    if isinstance(transport, dict):
        kind = transport.get("kind")
@@ -123,6 +135,11 @@ def _transport_from_connection_metadata(connection: ConnectionConfig) -> SourceT
                    str(key): str(value)
                    for key, value in dict(connection.metadata.get("env", {})).items()
                },
+                cwd=(
+                    str(connection.metadata["cwd"])
+                    if connection.metadata.get("cwd") is not None
+                    else None
+                ),
            )
        if transport in _FLAT_HTTP_TRANSPORTS:
            url = connection.metadata.get("url", "")
@@ -138,7 +155,7 @@ def _transport_from_connection_metadata(connection: ConnectionConfig) -> SourceT
        raise ValueError(
            f"connection {connection.id!r} has unrecognized metadata.transport {transport!r}"
        )
-    raise ValueError(f"connection {connection.id!r} requires metadata.transport")
+    return None


 __all__ = [
@@ -147,6 +147,7 @@ def connection_config_to_registry_entry(
                "command": connection.metadata.get("command", ""),
                "args": list(connection.metadata.get("args", [])),
                "env": dict(connection.metadata.get("env", {})),
+                "cwd": connection.metadata.get("cwd"),
            }
        elif transport in _FLAT_HTTP_TRANSPORTS:
            legacy_transport_value = transport
@@ -12,6 +12,7 @@ class StdioSourceTransport(SourceRegistryBaseModel):
    command: str = Field(min_length=1)
    args: tuple[str, ...] = ()
    env: dict[str, str] = Field(default_factory=dict)
+    cwd: str | None = None


 class HttpSourceTransport(SourceRegistryBaseModel):
@@ -122,6 +122,7 @@ def test_mcp_source_connection_from_legacy_connection_config_stdio() -> None:
            "command": "uvx",
            "args": ["github-mcp"],
            "env": {"A": "B"},
+            "cwd": "C:/repo",
            "auth_ref": "github.token",
            "profile": "engineering",
            "source_registry": True,
@@ -141,6 +142,7 @@ def test_mcp_source_connection_from_legacy_connection_config_stdio() -> None:
    assert isinstance(connection.transport, StdioSourceTransport)
    assert connection.transport.command == "uvx"
    assert connection.transport.args == ("github-mcp",)
+    assert connection.transport.cwd == "C:/repo"


 def test_mcp_source_connection_from_legacy_connection_config_http() -> None:
@@ -164,7 +166,7 @@ def test_mcp_source_connection_from_legacy_connection_config_http() -> None:
    assert connection.transport.headers == {"X-Test": "yes"}


-def test_mcp_source_connection_rejects_missing_legacy_transport() -> None:
+def test_mcp_source_connection_accepts_missing_legacy_transport_until_open() -> None:
    from wf_mcp.broker.models import ConnectionConfig

    legacy = ConnectionConfig(
@@ -174,8 +176,9 @@ def test_mcp_source_connection_rejects_missing_legacy_transport() -> None:
        metadata={},
    )

-    with pytest.raises(ValueError, match="requires metadata.transport"):
-        mcp_source_connection_from_connection_config(legacy)
+    connection = mcp_source_connection_from_connection_config(legacy)
+
+    assert connection.transport is None


 class _ConnectionLike(Protocol):