Repository: agentscope-ai/CoPaw (python)

Issue:
[Bug]:  对话报错Unknown agent error: ClosedResourceError:/tmp/copaw_query_error_w5fy9zl4.json

/tmp/copaw_query_error_w5fy9zl4.json内容如下：
```shell
# cat /tmp/copaw_query_error_w5fy9zl4.json
{
  "trace": "Traceback (most recent call last):\n  File \"/app/venv/lib/python3.11/site-packages/copaw/app/runner/runner.py\", line 106, in query_handler\n    await agent.register_mcp_clients()\n  File \"/app/venv/lib/python3.11/site-packages/copaw/agents/react_agent.py\", line 275, in register_mcp_clients\n    await self.toolkit.register_mcp_client(client)\n  File \"/app/venv/lib/python3.11/site-packages/agentscope/tool/_toolkit.py\", line 857, in register_mcp_client\n    for mcp_tool in await mcp_client.list_tools():\n                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/app/venv/lib/python3.11/site-packages/agentscope/mcp/_stateful_client_base.py\", line 106, in list_tools\n    res = await self.session.list_tools()\n          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/app/venv/lib/python3.11/site-packages/mcp/client/session.py\", line 529, in list_tools\n    result = await self.send_request(\n             ^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/app/venv/lib/python3.11/site-packages/mcp/shared/session.py\", line 281, in send_request\n    await self._write_stream.send(SessionMessage(message=JSONRPCMessage(jsonrpc_request), metadata=metadata))\n  File \"/app/venv/lib/python3.11/site-packages/anyio/streams/memory.py\", line 249, in send\n    self.send_nowait(item)\n  File \"/app/venv/lib/python3.11/site-packages/anyio/streams/memory.py\", line 218, in send_nowait\n    raise ClosedResourceError\nanyio.ClosedResourceError\n",
  "exception_type": "ClosedResourceError",
  "exception_message": "",
  "request_info": {
    "session_id": "1772499692494",
    "user_id": "default",
    "channel": "console"
  },
  "request": {
    "input": [
      {
        "sequence_number": null,
        "object": "message",
        "status": "created",
        "error": null,
        "id": "msg_008d58ff-b809-4a12-8fd6-fd76f55d6492",
        "type": "message",
        "role": "user",
        "content": [
          {
            "sequence_number": null,
            "object": "content",
            "status": "created",
            "error": null,
            "type": "text",
            "index": null,
            "delta": false,
            "msg_id": null,
            "text": "你好"
          }
        ],
        "code": null,
        "message": null,
        "usage": null,
        "metadata": null
      }
    ],
    "stream": true,
    "id": null,
    "model": null,
    "top_p": null,
    "temperature": null,
    "frequency_penalty": null,
    "presence_penalty": null,
    "max_tokens": null,
    "stop": null,
    "n": 1,
    "seed": null,
    "tools": null,
    "session_id": "1772499692494",
    "user_id": "default",
    "channel": "console"
  },
  "agent_state": {
    "memory": {
      "content": [],
      "_compressed_summary": ""
    },
    "toolkit": {
      "active_groups": []
    },
    "name": "Friday",
    "_sys_prompt": "====================\n- 当前的session_id: 1772499692494\n- 当前的user_id: default\n- 当前的channel: console\n- 工作目录: /app/working\n- ���要提示:\n  1. 完成任务时，优先考虑使用 skills（例如定时任务，优先使用 cron skill）。对于不清楚的 skills，请先查阅相关对应文档。\n  2. 使用 write_file 写文件时，如果担心覆盖原有内容，可以先用 read_file 查看文件内容，再使用 edit_file 工具进行局部内容更新或追加内容。\n====================\n\n\nYou are a helpful assistant.\n"
  },
  "ts_utc": "2026-03-03T02:44:23.268054Z"
}
```

复现过程：
执行复杂对话后，等待一段时间，再次对话就会这样，此时会话页面无响应了，重启才恢复，但是对话记录都没了

CoPaw版本：0.0.4
1panel 应用商店安装


Patch:
diff --git a/src/copaw/agents/react_agent.py b/src/copaw/agents/react_agent.py
index 758d3c9b1..45a7d1d68 100644
--- a/src/copaw/agents/react_agent.py
+++ b/src/copaw/agents/react_agent.py
@@ -4,12 +4,14 @@
 This module provides the main CoPawAgent class built on ReActAgent,
 with integrated tools, skills, and memory management.
 """
+import asyncio
 import logging
 import os
 from typing import Any, List, Literal, Optional, Type
 
-
+from anyio import ClosedResourceError
 from agentscope.agent import ReActAgent
+from agentscope.mcp import HttpStatefulClient, StdIOStatefulClient
 from agentscope.message import Msg
 from agentscope.tool import Toolkit
 from pydantic import BaseModel
@@ -327,11 +329,170 @@ async def register_mcp_clients(
                 Options: "override", "skip", "raise", "rename"
                 (default: "skip")
         """
-        for client in self._mcp_clients:
-            await self.toolkit.register_mcp_client(
-                client,
-                namesake_strategy=namesake_strategy,
+        for i, client in enumerate(self._mcp_clients):
+            client_name = getattr(client, "name", repr(client))
+            try:
+                await self.toolkit.register_mcp_client(
+                    client,
+                    namesake_strategy=namesake_strategy,
+                )
+            except (ClosedResourceError, asyncio.CancelledError) as error:
+                if self._should_propagate_cancelled_error(error):
+                    raise
+                logger.warning(
+                    "MCP client '%s' session interrupted while listing tools; "
+                    "trying recovery",
+                    client_name,
+                )
+                recovered_client = await self._recover_mcp_client(client)
+                if recovered_client is not None:
+                    self._mcp_clients[i] = recovered_client
+                    try:
+                        await self.toolkit.register_mcp_client(
+                            recovered_client,
+                            namesake_strategy=namesake_strategy,
+                        )
+                        continue
+                    except asyncio.CancelledError as recover_error:
+                        if self._should_propagate_cancelled_error(
+                            recover_error,
+                        ):
+                            raise
+                        logger.warning(
+                            "MCP client '%s' registration cancelled after "
+                            "recovery, skipping",
+                            client_name,
+                        )
+                    except Exception as e:  # pylint: disable=broad-except
+                        logger.warning(
+                            "MCP client '%s' still unavailable after "
+                            "recovery, skipping: %s",
+                            client_name,
+                            e,
+                        )
+                else:
+                    logger.warning(
+                        "MCP client '%s' recovery failed, skipping",
+                        client_name,
+                    )
+            except Exception as e:  # pylint: disable=broad-except
+                logger.exception(
+                    "Unexpected error registering MCP client '%s': %s",
+                    client_name,
+                    e,
+                )
+                raise
+
+    async def _recover_mcp_client(self, client: Any) -> Any | None:
+        """Recover MCP client from broken session and return healthy client."""
+        if await self._reconnect_mcp_client(client):
+            return client
+
+        rebuilt_client = self._rebuild_mcp_client(client)
+        if rebuilt_client is None:
+            return None
+
+        if await self._reconnect_mcp_client(rebuilt_client):
+            return self._reuse_shared_client_reference(
+                original_client=client,
+                rebuilt_client=rebuilt_client,
+            )
+
+        return None
+
+    @staticmethod
+    def _reuse_shared_client_reference(
+        original_client: Any,
+        rebuilt_client: Any,
+    ) -> Any:
+        """Keep manager-shared client reference stable after rebuild."""
+        original_dict = getattr(original_client, "__dict__", None)
+        rebuilt_dict = getattr(rebuilt_client, "__dict__", None)
+        if isinstance(original_dict, dict) and isinstance(rebuilt_dict, dict):
+            original_dict.update(rebuilt_dict)
+            return original_client
+        return rebuilt_client
+
+    @staticmethod
+    def _should_propagate_cancelled_error(error: BaseException) -> bool:
+        """Only swallow MCP-internal cancellations, not task cancellation."""
+        if not isinstance(error, asyncio.CancelledError):
+            return False
+
+        task = asyncio.current_task()
+        if task is None:
+            return False
+
+        cancelling = getattr(task, "cancelling", None)
+        if callable(cancelling):
+            return cancelling() > 0
+
+        # Python < 3.11: Task.cancelling() is unavailable.
+        # Fall back to propagating CancelledError to avoid swallowing
+        # genuine task cancellations when we cannot inspect the state.
+        return True
+
+    @staticmethod
+    async def _reconnect_mcp_client(
+        client: Any,
+        timeout: float = 60.0,
+    ) -> bool:
+        """Best-effort reconnect for stateful MCP clients."""
+        close_fn = getattr(client, "close", None)
+        if callable(close_fn):
+            try:
+                await close_fn()
+            except asyncio.CancelledError:  # pylint: disable=try-except-raise
+                raise
+            except Exception:  # pylint: disable=broad-except
+                pass
+
+        connect_fn = getattr(client, "connect", None)
+        if not callable(connect_fn):
+            return False
+
+        try:
+            await asyncio.wait_for(connect_fn(), timeout=timeout)
+            return True
+        except asyncio.CancelledError:  # pylint: disable=try-except-raise
+            raise
+        except asyncio.TimeoutError:
+            return False
+        except Exception:  # pylint: disable=broad-except
+            return False
+
+    @staticmethod
+    def _rebuild_mcp_client(client: Any) -> Any | None:
+        """Rebuild a fresh MCP client instance from stored config metadata."""
+        rebuild_info = getattr(client, "_copaw_rebuild_info", None)
+        if not isinstance(rebuild_info, dict):
+            return None
+
+        transport = rebuild_info.get("transport")
+        name = rebuild_info.get("name")
+
+        try:
+            if transport == "stdio":
+                rebuilt_client = StdIOStatefulClient(
+                    name=name,
+                    command=rebuild_info.get("command"),
+                    args=rebuild_info.get("args", []),
+                    env=rebuild_info.get("env", {}),
+                    cwd=rebuild_info.get("cwd"),
+                )
+                setattr(rebuilt_client, "_copaw_rebuild_info", rebuild_info)
+                return rebuilt_client
+
+            rebuilt_client = HttpStatefulClient(
+                name=name,
+                transport=transport,
+                url=rebuild_info.get("url"),
+                headers=rebuild_info.get("headers"),
             )
+            setattr(rebuilt_client, "_copaw_rebuild_info", rebuild_info)
+            return rebuilt_client
+        except Exception:  # pylint: disable=broad-except
+            return None
 
     async def _reasoning(
         self,
diff --git a/src/copaw/app/mcp/manager.py b/src/copaw/app/mcp/manager.py
index 4cd8bcec8..69f9fdb15 100644
--- a/src/copaw/app/mcp/manager.py
+++ b/src/copaw/app/mcp/manager.py
@@ -185,18 +185,33 @@ async def _add_client(
     @staticmethod
     def _build_client(client_config: "MCPClientConfig") -> Any:
         """Build MCP client instance by configured transport."""
+        rebuild_info = {
+            "name": client_config.name,
+            "transport": client_config.transport,
+            "url": client_config.url,
+            "headers": client_config.headers or None,
+            "command": client_config.command,
+            "args": list(client_config.args),
+            "env": dict(client_config.env),
+            "cwd": client_config.cwd or None,
+        }
+
         if client_config.transport == "stdio":
-            return StdIOStatefulClient(
+            client = StdIOStatefulClient(
                 name=client_config.name,
                 command=client_config.command,
                 args=client_config.args,
                 env=client_config.env,
                 cwd=client_config.cwd or None,
             )
+            setattr(client, "_copaw_rebuild_info", rebuild_info)
+            return client
 
-        return HttpStatefulClient(
+        client = HttpStatefulClient(
             name=client_config.name,
             transport=client_config.transport,
             url=client_config.url,
             headers=client_config.headers or None,
         )
+        setattr(client, "_copaw_rebuild_info", rebuild_info)
+        return client
diff --git a/src/copaw/app/runner/runner.py b/src/copaw/app/runner/runner.py
index 93017980d..c1d56c184 100644
--- a/src/copaw/app/runner/runner.py
+++ b/src/copaw/app/runner/runner.py
@@ -59,6 +59,7 @@ async def query_handler(
 
         agent = None
         chat = None
+        session_state_loaded = False
 
         try:
             session_id = request.session_id
@@ -131,6 +132,7 @@ async def query_handler(
                 user_id=user_id,
                 agent=agent,
             )
+            session_state_loaded = True
 
             # Rebuild system prompt so it always reflects the latest
             # AGENTS.md / SOUL.md / PROFILE.md, not the stale one saved
@@ -169,7 +171,7 @@ async def query_handler(
                 ) + e.args[1:]
             raise
         finally:
-            if agent is not None:
+            if agent is not None and session_state_loaded:
                 await self.session.save_session_state(
                     session_id=session_id,
                     user_id=user_id,


Test patch:
diff --git a/tests/test_mcp_resilience.py b/tests/test_mcp_resilience.py
new file mode 100644
index 000000000..033e618cc
--- /dev/null
+++ b/tests/test_mcp_resilience.py
@@ -0,0 +1,273 @@
+# -*- coding: utf-8 -*-
+# pylint: disable=protected-access,unused-argument
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from anyio import ClosedResourceError
+
+import copaw.app.runner.runner as runner_module
+from copaw.agents.react_agent import CoPawAgent
+from copaw.app.mcp.manager import MCPClientManager
+from copaw.app.runner.runner import AgentRunner
+from copaw.config.config import MCPClientConfig
+
+
+class _FakeToolkit:
+    def __init__(
+        self,
+        fail_once_names: set[str] | None = None,
+        always_fail_names: set[str] | None = None,
+        runtime_fail_names: set[str] | None = None,
+    ) -> None:
+        self.fail_once_names = fail_once_names or set()
+        self.always_fail_names = always_fail_names or set()
+        self.runtime_fail_names = runtime_fail_names or set()
+        self.calls: dict[str, int] = {}
+        self.registered: list[str] = []
+        self.cancel_once_names: set[str] = set()
+
+    async def register_mcp_client(
+        self,
+        client,
+        namesake_strategy: str = "skip",  # noqa: ARG002
+    ) -> None:
+        name = client.name
+        self.calls[name] = self.calls.get(name, 0) + 1
+
+        if name in self.always_fail_names:
+            raise ClosedResourceError()
+
+        if name in self.runtime_fail_names:
+            raise RuntimeError("unexpected toolkit failure")
+
+        if name in self.cancel_once_names and self.calls[name] == 1:
+            raise asyncio.CancelledError()
+
+        if name in self.fail_once_names and self.calls[name] == 1:
+            raise ClosedResourceError()
+
+        self.registered.append(name)
+
+
+class _FakeMCPClient:
+    def __init__(self, name: str, connect_ok: bool = True) -> None:
+        self.name = name
+        self.connect_ok = connect_ok
+        self.close_calls = 0
+        self.connect_calls = 0
+
+    async def close(self) -> None:
+        self.close_calls += 1
+
+    async def connect(self) -> None:
+        self.connect_calls += 1
+        if not self.connect_ok:
+            raise RuntimeError("connect failed")
+
+
+def test_build_client_attaches_rebuild_info(tmp_path: Path) -> None:
+    cfg = MCPClientConfig(
+        name="mcp_everything",
+        enabled=True,
+        transport="stdio",
+        command="npx",
+        args=["-y", "@modelcontextprotocol/server-everything"],
+        env={"A": "1"},
+        cwd=str(tmp_path),
+    )
+
+    client = MCPClientManager._build_client(cfg)
+    rebuild_info = getattr(client, "_copaw_rebuild_info", None)
+
+    assert isinstance(rebuild_info, dict)
+    assert rebuild_info["transport"] == "stdio"
+    assert rebuild_info["command"] == "npx"
+    assert rebuild_info["args"] == [
+        "-y",
+        "@modelcontextprotocol/server-everything",
+    ]
+    assert rebuild_info["env"] == {"A": "1"}
+    assert rebuild_info["cwd"] == str(tmp_path)
+
+
+@pytest.mark.asyncio
+async def test_register_mcp_clients_retries_once_on_closed_resource() -> None:
+    toolkit = _FakeToolkit(fail_once_names={"flaky"})
+    flaky = _FakeMCPClient(name="flaky", connect_ok=True)
+    healthy = _FakeMCPClient(name="healthy", connect_ok=True)
+
+    agent = object.__new__(CoPawAgent)
+    agent.toolkit = toolkit
+    agent._mcp_clients = [flaky, healthy]
+
+    await CoPawAgent.register_mcp_clients(agent)
+
+    assert toolkit.calls["flaky"] == 2
+    assert flaky.connect_calls == 1
+    assert toolkit.calls["healthy"] == 1
+    assert toolkit.registered == ["flaky", "healthy"]
+
+
+@pytest.mark.asyncio
+async def test_register_mcp_clients_skips_unrecoverable_client() -> None:
+    toolkit = _FakeToolkit(always_fail_names={"broken"})
+    broken = _FakeMCPClient(name="broken", connect_ok=False)
+    healthy = _FakeMCPClient(name="healthy", connect_ok=True)
+
+    agent = object.__new__(CoPawAgent)
+    agent.toolkit = toolkit
+    agent._mcp_clients = [broken, healthy]
+
+    await CoPawAgent.register_mcp_clients(agent)
+
+    assert toolkit.calls["broken"] == 1
+    assert broken.connect_calls == 1
+    assert "broken" not in toolkit.registered
+    assert toolkit.registered == ["healthy"]
+
+
+@pytest.mark.asyncio
+async def test_register_mcp_clients_handles_cancelled_error() -> None:
+    toolkit = _FakeToolkit()
+    toolkit.cancel_once_names = {"flaky"}
+    flaky = _FakeMCPClient(name="flaky", connect_ok=True)
+
+    agent = object.__new__(CoPawAgent)
+    agent.toolkit = toolkit
+    agent._mcp_clients = [flaky]
+
+    await CoPawAgent.register_mcp_clients(agent)
+
+    assert toolkit.calls["flaky"] == 2
+    assert flaky.connect_calls == 1
+    assert toolkit.registered == ["flaky"]
+
+
+@pytest.mark.asyncio
+async def test_register_mcp_clients_reraises_unexpected_error() -> None:
+    toolkit = _FakeToolkit(runtime_fail_names={"boom"})
+    boom = _FakeMCPClient(name="boom", connect_ok=True)
+
+    agent = object.__new__(CoPawAgent)
+    agent.toolkit = toolkit
+    agent._mcp_clients = [boom]
+
+    with pytest.raises(RuntimeError, match="unexpected toolkit failure"):
+        await CoPawAgent.register_mcp_clients(agent)
+
+
+@pytest.mark.asyncio
+async def test_register_mcp_clients_rebuilds_client_when_reconnect_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    toolkit = _FakeToolkit(always_fail_names={"broken"})
+    broken = _FakeMCPClient(name="broken", connect_ok=False)
+    rebuilt = _FakeMCPClient(name="rebuilt", connect_ok=True)
+
+    monkeypatch.setattr(
+        CoPawAgent,
+        "_rebuild_mcp_client",
+        staticmethod(lambda client: rebuilt),  # noqa: ARG005
+    )
+
+    agent = object.__new__(CoPawAgent)
+    agent.toolkit = toolkit
+    agent._mcp_clients = [broken]
+
+    await CoPawAgent.register_mcp_clients(agent)
+
+    assert broken.connect_calls == 1
+    assert rebuilt.connect_calls == 1
+    assert toolkit.registered == ["rebuilt"]
+    assert agent._mcp_clients[0] is broken
+    assert agent._mcp_clients[0].name == "rebuilt"
+
+
+@pytest.mark.asyncio
+async def test_reconnect_mcp_client_respects_timeout() -> None:
+    class _SlowClient:
+        async def close(self) -> None:
+            return
+
+        async def connect(self) -> None:
+            await asyncio.sleep(0.1)
+
+    ok = await CoPawAgent._reconnect_mcp_client(
+        _SlowClient(),
+        timeout=0.01,
+    )
+    assert ok is False
+
+
+@pytest.mark.asyncio
+async def test_query_handler_skips_session_save_when_load_not_reached(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class _FakeAgent:
+        def __init__(self, *args, **kwargs) -> None:  # noqa: ARG002
+            pass
+
+        async def register_mcp_clients(self) -> None:
+            raise ClosedResourceError()
+
+        def set_console_output_enabled(
+            self,
+            enabled: bool,
+        ) -> None:  # noqa: ARG002
+            return
+
+    class _FakeSession:
+        def __init__(self) -> None:
+            self.load_calls = 0
+            self.save_calls = 0
+
+        async def load_session_state(self, **kwargs) -> None:  # noqa: ARG002
+            self.load_calls += 1
+
+        async def save_session_state(self, **kwargs) -> None:  # noqa: ARG002
+            self.save_calls += 1
+
+    class _DummyInputMsg:
+        def get_text_content(self) -> str:
+            return "你好"
+
+    cfg = SimpleNamespace(
+        agents=SimpleNamespace(
+            running=SimpleNamespace(max_iters=1, max_input_length=2048),
+        ),
+    )
+
+    monkeypatch.setattr(runner_module, "CoPawAgent", _FakeAgent)
+    monkeypatch.setattr(runner_module, "load_config", lambda: cfg)
+    monkeypatch.setattr(
+        runner_module,
+        "build_env_context",
+        lambda **kwargs: "env",
+    )
+    monkeypatch.setattr(
+        runner_module,
+        "write_query_error_dump",
+        lambda **kwargs: None,
+    )
+
+    runner = AgentRunner()
+    fake_session = _FakeSession()
+    runner.session = fake_session
+
+    request = SimpleNamespace(
+        session_id="s1",
+        user_id="u1",
+        channel="console",
+    )
+
+    with pytest.raises(ClosedResourceError):
+        async for _ in runner.query_handler(
+            [_DummyInputMsg()],
+            request=request,
+        ):
+            pass
+
+    assert fake_session.load_calls == 0
+    assert fake_session.save_calls == 0