fix: 优化外部资源生命周期和 MCP streamable-http 兼容

- Runner 支持通过 close_session_service_on_close /
  close_memory_service_on_close 控制是否在 close 时关闭外部托管的
session_service 和 memory_service。
- MCP Streamable HTTP client 优先使用新的 streamable_http_client
  API，并在旧版 MCP SDK 下回退到 streamablehttp_client。
- 使用 httpx.AsyncClient 承载 streamable-http 的 headers / timeout
  配置，避免调用已废弃接口。
- 补充 Runner 和 MCP session manager 相关测试。

Author: raychen911

pyproject.toml

@@ -26,7 +26,7 @@ classifiers = [
 dependencies = [
     "pydantic>=2.11.3",
     "openai>=1.3.0",
-    "mcp<1.23.4,>=1.10.1",
+    "mcp>=1.10.1",
     "aiohttp",
     "httpx>=0.27.0",
     "httpx-sse>=0.4.0",

tests/test_runner.py

@@ -107,6 +107,8 @@ def test_init_with_required_params(self, mock_agent, mock_session_service):
         assert runner.session_service == mock_session_service
         assert runner.artifact_service is None
         assert runner.memory_service is None
+        assert runner._close_session_service_on_close is True
+        assert runner._close_memory_service_on_close is True
 
     def test_init_with_all_params(self, mock_agent, mock_session_service):
         """Test Runner initialization with all parameters."""
@@ -119,10 +121,14 @@ def test_init_with_all_params(self, mock_agent, mock_session_service):
             session_service=mock_session_service,
             artifact_service=artifact_service,
             memory_service=memory_service,
+            close_session_service_on_close=False,
+            close_memory_service_on_close=False,
         )
 
         assert runner.artifact_service == artifact_service
         assert runner.memory_service == memory_service
+        assert runner._close_session_service_on_close is False
+        assert runner._close_memory_service_on_close is False
 
 
 # Tests for run_async method
@@ -618,6 +624,25 @@ async def test_close_calls_all_services(self, runner, mock_session_service):
         mock_session_service.close.assert_called_once()
         memory_service.close.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_close_can_keep_external_services_open(self, mock_agent, mock_session_service):
+        """Test close skips externally managed session and memory services."""
+        memory_service = AsyncMock(spec=BaseMemoryService)
+        runner = Runner(
+            app_name="test_app",
+            agent=mock_agent,
+            session_service=mock_session_service,
+            memory_service=memory_service,
+            close_session_service_on_close=False,
+            close_memory_service_on_close=False,
+        )
+
+        with patch.object(runner, '_collect_toolset', return_value=set()):
+            await runner.close()
+
+        mock_session_service.close.assert_not_called()
+        memory_service.close.assert_not_called()
+
 
 # Tests for session history handling
 class TestSessionHistory:

tests/tools/mcp_tool/test_mcp_session_manager.py

@@ -233,11 +233,11 @@ def test_sse_client(self, mock_sse):
         call_kwargs = mock_sse.call_args
         assert call_kwargs.kwargs["headers"] == {"Auth": "token"}
 
-    @patch("trpc_agent_sdk.tools.mcp_tool._mcp_session_manager.streamablehttp_client")
+    @patch.object(MCPSessionManager, "_create_streamable_http_client")
     def test_streamable_client(self, mock_streamable):
         mgr = MCPSessionManager(connection_params=_streamable_conn())
         mgr._create_client({"Auth": "token"})
-        mock_streamable.assert_called_once()
+        mock_streamable.assert_called_once_with({"Auth": "token"})
 
     def test_unsupported_params_raises(self):
         mgr = MCPSessionManager(connection_params=_stdio_conn())

trpc_agent_sdk/runners.py

@@ -194,6 +194,8 @@ def __init__(
         enable_post_turn_processing: bool = True,
         defer_post_turn_processing: bool = False,
         post_turn_queue_maxsize: int = 256,
+        close_session_service_on_close: bool = True,
+        close_memory_service_on_close: bool = True,
     ):
         """Initializes the Runner.
 
@@ -209,6 +211,13 @@ def __init__(
                 persistence run in a dedicated thread/event loop so request
                 completion is not blocked by post-turn I/O/LLM latency.
             post_turn_queue_maxsize: Max buffered post-turn jobs per runner.
+            close_session_service_on_close: Whether :meth:`close` should close
+                the session service. Set to False when the service is managed
+                outside the runner, e.g. an application-level Redis connection
+                pool shared by many short-lived runners.
+            close_memory_service_on_close: Whether :meth:`close` should close
+                the memory service. Set to False when the service is managed
+                outside the runner.
         """
         self.app_name = app_name
         self.agent = agent
@@ -219,6 +228,8 @@ def __init__(
         self._defer_post_turn_processing = defer_post_turn_processing
         self._post_turn_thread: _PostTurnWorkerThread | None = None
         self._post_turn_queue_maxsize = max(1, int(post_turn_queue_maxsize))
+        self._close_session_service_on_close = close_session_service_on_close
+        self._close_memory_service_on_close = close_memory_service_on_close
 
     async def _run_post_turn_processing(
         self,
@@ -823,7 +834,7 @@ async def close(self):
         """
         await self._shutdown_post_turn_worker()
         await self._cleanup_toolsets(self._collect_toolset(self.agent))
-        if self.session_service:
+        if self.session_service and self._close_session_service_on_close:
             await self.session_service.close()
-        if self.memory_service:
+        if self.memory_service and self._close_memory_service_on_close:
             await self.memory_service.close()

trpc_agent_sdk/tools/mcp_tool/_mcp_session_manager.py

@@ -26,18 +26,28 @@
 
 import asyncio
 import hashlib
+import inspect
 import json
 import sys
 from contextlib import AsyncExitStack
+from contextlib import asynccontextmanager
 from datetime import timedelta
 from typing import Dict
 from typing import Optional
 from typing import Union
 
+# cSpell:ignore streamable
+
+import httpx
 from mcp import ClientSession
 from mcp.client.sse import sse_client
 from mcp.client.stdio import stdio_client
-from mcp.client.streamable_http import streamablehttp_client
+try:
+    from mcp.client.streamable_http import streamable_http_client
+except ImportError:  # pragma: no cover - compatibility with older mcp versions
+    from mcp.client.streamable_http import streamablehttp_client as streamable_http_client
+
+_STREAMABLE_HTTP_CLIENT_SUPPORTS_HTTP_CLIENT = "http_client" in inspect.signature(streamable_http_client).parameters
 
 from trpc_agent_sdk.log import logger
 
@@ -161,7 +171,7 @@ def _create_client(self, merged_headers: Optional[Dict[str, str]] = None):
         if isinstance(self._connection_params, StdioConnectionParams):
             client = stdio_client(
                 server=self._connection_params.server_params,
-                errlog=sys.stderr,
+                errlog=sys.stderr,  # cSpell:ignore errlog
             )
         elif isinstance(self._connection_params, SseConnectionParams):
             client = sse_client(
@@ -171,18 +181,46 @@ def _create_client(self, merged_headers: Optional[Dict[str, str]] = None):
                 sse_read_timeout=self._connection_params.sse_read_timeout,
             )
         elif isinstance(self._connection_params, StreamableHTTPConnectionParams):
-            client = streamablehttp_client(
+            client = self._create_streamable_http_client(merged_headers)
+        else:
+            raise ValueError('Unable to initialize connection. Connection should be'
+                             ' StdioConnectionParams or SseConnectionParams or StreamableHTTPConnectionParams, but got'
+                             f' {type(self._connection_params)}')
+        return client
+
+    def _create_streamable_http_client(self, merged_headers: Optional[Dict[str, str]] = None):
+        """Creates a Streamable HTTP client with compatibility for MCP SDK versions."""
+        if not _STREAMABLE_HTTP_CLIENT_SUPPORTS_HTTP_CLIENT:
+            return streamable_http_client(
                 url=self._connection_params.url,
                 headers=merged_headers,
                 timeout=self._connection_params.timeout,
                 sse_read_timeout=self._connection_params.sse_read_timeout,
                 terminate_on_close=self._connection_params.terminate_on_close,
             )
-        else:
-            raise ValueError('Unable to initialize connection. Connection should be'
-                             ' StdioConnectionParams or SseConnectionParams or StreamableHTTPConnectionParams, but got'
-                             f' {type(self._connection_params)}')
-        return client
+
+        return self._create_streamable_http_client_with_httpx(merged_headers)
+
+    @asynccontextmanager
+    async def _create_streamable_http_client_with_httpx(self, merged_headers: Optional[Dict[str, str]] = None):
+        """Creates a new-style Streamable HTTP client and owns its httpx client."""
+        timeout_seconds = self._connection_params.timeout
+        if isinstance(timeout_seconds, timedelta):
+            timeout_seconds = timeout_seconds.total_seconds()
+        sse_read_timeout_seconds = self._connection_params.sse_read_timeout
+        if isinstance(sse_read_timeout_seconds, timedelta):
+            sse_read_timeout_seconds = sse_read_timeout_seconds.total_seconds()
+
+        async with httpx.AsyncClient(
+                headers=merged_headers,
+                timeout=httpx.Timeout(timeout_seconds, read=sse_read_timeout_seconds),
+        ) as http_client:
+            async with streamable_http_client(
+                    url=self._connection_params.url,
+                    http_client=http_client,
+                    terminate_on_close=self._connection_params.terminate_on_close,
+            ) as transports:
+                yield transports
 
     async def create_session(self, headers: Optional[Dict[str, str]] = None) -> ClientSession | None:
         """Creates and initializes an MCP client session.