Skip to content

refactor(mcp): migrate from MCP SDK v1 to v2.0.0b1 #112

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jonpspri wants to merge 5 commits into
base: 0.1.x
Choose a base branch
from
Draft

refactor(mcp): migrate from MCP SDK v1 to v2.0.0b1 #112

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a8fa6db
refactor(mcp): migrate from MCP SDK v1 to v2.0.0b1
jonpspri Jul 3, 2026
7a72951
chore(mcp): apply review fixes from PR #112
araujof Jul 13, 2026
58acdbb
fix(mcp): pass bind host to streamable_http_app and close caller-owne…
araujof Jul 13, 2026
eac9437
test(mcp): cover v2 reconnect, transport wiring, and session security
araujof Jul 13, 2026
8078691
Upgrade mcp SDK from 2.0.0b1 to 2.0.0b2
jonpspri Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ stats/
*cookies*txt
cookies*
cookies.txt
.bob
.claude
mutants
.mutmut-cache
Expand Down Expand Up @@ -250,4 +251,4 @@ db_path/
tmp/

.continue
plugin-catalog
plugin-catalog
524 changes: 524 additions & 0 deletions MIGRATION_MCP_V2.md
View file
Open in desktop

Large diffs are not rendered by default.

75 changes: 24 additions & 51 deletions cpex/framework/external/mcp/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,12 +19,12 @@
from typing import Any, Awaitable, Callable, Optional

# Third-Party
import httpx
import httpx2
import orjson
from mcp import ClientSession, McpError, StdioServerParameters
from mcp import ClientSession, MCPError, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.streamable_http import streamablehttp_client
from mcp.types import TextContent
from mcp.client.streamable_http import streamable_http_client
from mcp_types import TextContent

# First-Party
from cpex.framework.base import HookRef, Plugin, PluginRef
Expand Down Expand Up @@ -82,9 +82,6 @@ def __init__(self, config: PluginConfig) -> None:
self._stdio_ready: Optional[asyncio.Event] = None
self._stdio_stop: Optional[asyncio.Event] = None
self._stdio_error: Optional[BaseException] = None
self._get_session_id: Optional[Callable[[], str | None]] = None
self._session_id: Optional[str] = None
self._http_client_factory: Optional[Callable[..., httpx.AsyncClient]] = None
self._reconnect_attempts: int = 3
self._reconnect_delay: float = 0.1
self._reconnect_lock: asyncio.Lock = asyncio.Lock()
Expand Down Expand Up @@ -318,10 +315,10 @@ async def __connect_to_http_server(self, uri: str) -> None:

def _tls_httpx_client_factory(
headers: Optional[dict[str, str]] = None,
timeout: Optional[httpx.Timeout] = None,
auth: Optional[httpx.Auth] = None,
) -> httpx.AsyncClient:
"""Build an httpx client with TLS configuration for external MCP servers.
timeout: Optional[httpx2.Timeout] = None,
auth: Optional[httpx2.Auth] = None,
) -> httpx2.AsyncClient:
"""Build an httpx2 client with TLS configuration for external MCP servers.

Args:
headers: Optional HTTP headers to include in requests.
Expand All @@ -337,14 +334,14 @@ def _tls_httpx_client_factory(

kwargs: dict[str, Any] = {"follow_redirects": True}
if uds_path:
kwargs["transport"] = httpx.AsyncHTTPTransport(uds=uds_path)
kwargs["transport"] = httpx2.AsyncHTTPTransport(uds=uds_path)
if headers:
kwargs["headers"] = headers
http_settings = get_http_client_settings()
kwargs["timeout"] = (
timeout
if timeout
else httpx.Timeout(
else httpx2.Timeout(
connect=http_settings.httpx_connect_timeout,
read=http_settings.httpx_read_timeout,
write=http_settings.httpx_write_timeout,
Expand All @@ -355,7 +352,7 @@ def _tls_httpx_client_factory(
kwargs["auth"] = auth

# Add connection pool limits
kwargs["limits"] = httpx.Limits(
kwargs["limits"] = httpx2.Limits(
max_connections=http_settings.httpx_max_connections,
max_keepalive_connections=http_settings.httpx_max_keepalive_connections,
keepalive_expiry=http_settings.httpx_keepalive_expiry,
Expand All @@ -364,32 +361,34 @@ def _tls_httpx_client_factory(
if not tls_config:
# Use skip_ssl_verify setting when no custom TLS config
kwargs["verify"] = not http_settings.skip_ssl_verify
return httpx.AsyncClient(**kwargs)
return httpx2.AsyncClient(**kwargs)

# Create SSL context using the utility function
# This implements certificate validation per test_client_certificate_validation.py
ssl_context = create_ssl_context(tls_config, self.name)
kwargs["verify"] = ssl_context

return httpx.AsyncClient(**kwargs)
return httpx2.AsyncClient(**kwargs)

self._http_client_factory = _tls_httpx_client_factory
max_retries = 3
base_delay = 1.0

for attempt in range(max_retries):
try:
client_factory = _tls_httpx_client_factory
streamable_client = streamablehttp_client(
uri, httpx_client_factory=client_factory, terminate_on_close=False
http_client = _tls_httpx_client_factory()
# The v2 SDK does not close a caller-provided http_client, so we own its
# lifecycle. Enter it into the exit stack first (before the transport) so
# LIFO teardown closes the transport — firing the terminate_on_close DELETE
# while the client is still open — and then closes the client itself.
await self._exit_stack.enter_async_context(http_client)
streamable_client = streamable_http_client(
uri, http_client=http_client, terminate_on_close=True
)
http_transport = await self._exit_stack.enter_async_context(streamable_client)
self._http, self._write, get_session_id = http_transport
self._get_session_id = get_session_id
self._http, self._write = http_transport
self._session = await self._exit_stack.enter_async_context(ClientSession(self._http, self._write))

await self._session.initialize()
self._session_id = self._get_session_id() if self._get_session_id else None
response = await self._session.list_tools()
tools = response.tools
logger.info(
Expand Down Expand Up @@ -446,8 +445,6 @@ async def _cleanup_session(self) -> None:
self._http = None
self._write = None
self._stdio = None
self._get_session_id = None
self._session_id = None

async def _reconnect_session(self) -> None:
"""Tear down old session and reconnect to MCP server with linear backoff.
Expand Down Expand Up @@ -570,8 +567,8 @@ async def _execute_call() -> PluginResult:
) from reconn_err
logger.exception(pe)
raise
except McpError as e:
logger.warning("McpError for plugin %s: %s", self.name, e)
except MCPError as e:
logger.warning("MCPError for plugin %s: %s", self.name, e)
try:
async with self._reconnect_lock:
await self._reconnect_session()
Expand Down Expand Up @@ -637,30 +634,6 @@ async def shutdown(self) -> None:

if self._exit_stack:
await self._exit_stack.aclose()
if self._config and self._config.mcp and self._config.mcp.proto == TransportType.STREAMABLEHTTP:
await self.__terminate_http_session()
self._get_session_id = None
self._session_id = None
self._http_client_factory = None

async def __terminate_http_session(self) -> None:
"""Terminate streamable HTTP session explicitly to avoid lingering server state."""
if not self._session_id or not self._config or not self._config.mcp or not self._config.mcp.url:
return
# Third-Party
from mcp.server.streamable_http import MCP_SESSION_ID_HEADER # pylint: disable=import-outside-toplevel

client_factory = self._http_client_factory
try:
if client_factory:
client = client_factory()
else:
client = httpx.AsyncClient(follow_redirects=True)
async with client:
headers = {MCP_SESSION_ID_HEADER: self._session_id}
await client.delete(self._config.mcp.url, headers=headers)
except Exception as exc:
logger.debug("Failed to terminate streamable HTTP session: %s", exc)


class ExternalHookRef(HookRef):
Expand Down
Loading