mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-07 21:40:14 +08:00
Compare commits
4 Commits
codex/even
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3d739c795f | ||
|
|
2a7c02af8e | ||
|
|
56f2533951 | ||
|
|
afb0079116 |
20
.github/workflows/docker-image.yml
vendored
20
.github/workflows/docker-image.yml
vendored
@@ -71,20 +71,20 @@ jobs:
|
||||
echo "build_date=$build_date" >> $GITHUB_OUTPUT
|
||||
|
||||
- name: Set QEMU
|
||||
uses: docker/setup-qemu-action@v4.1.0
|
||||
uses: docker/setup-qemu-action@v4.2.0
|
||||
|
||||
- name: Set Docker Buildx
|
||||
uses: docker/setup-buildx-action@v4.1.0
|
||||
uses: docker/setup-buildx-action@v4.2.0
|
||||
|
||||
- name: Log in to DockerHub
|
||||
uses: docker/login-action@v4.2.0
|
||||
uses: docker/login-action@v4.4.0
|
||||
with:
|
||||
username: ${{ secrets.DOCKER_HUB_USERNAME }}
|
||||
password: ${{ secrets.DOCKER_HUB_PASSWORD }}
|
||||
|
||||
- name: Login to GitHub Container Registry
|
||||
if: env.HAS_GHCR_TOKEN == 'true'
|
||||
uses: docker/login-action@v4.2.0
|
||||
uses: docker/login-action@v4.4.0
|
||||
with:
|
||||
registry: ghcr.io
|
||||
username: ${{ env.GHCR_OWNER }}
|
||||
@@ -105,7 +105,7 @@ jobs:
|
||||
echo "EOF" >> $GITHUB_OUTPUT
|
||||
|
||||
- name: Build and Push Nightly Image
|
||||
uses: docker/build-push-action@v7.2.0
|
||||
uses: docker/build-push-action@v7.3.0
|
||||
with:
|
||||
context: .
|
||||
platforms: linux/amd64,linux/arm64
|
||||
@@ -171,27 +171,27 @@ jobs:
|
||||
cp -r dashboard/dist astrbot/dashboard/dist
|
||||
|
||||
- name: Set QEMU
|
||||
uses: docker/setup-qemu-action@v4.1.0
|
||||
uses: docker/setup-qemu-action@v4.2.0
|
||||
|
||||
- name: Set Docker Buildx
|
||||
uses: docker/setup-buildx-action@v4.1.0
|
||||
uses: docker/setup-buildx-action@v4.2.0
|
||||
|
||||
- name: Log in to DockerHub
|
||||
uses: docker/login-action@v4.2.0
|
||||
uses: docker/login-action@v4.4.0
|
||||
with:
|
||||
username: ${{ secrets.DOCKER_HUB_USERNAME }}
|
||||
password: ${{ secrets.DOCKER_HUB_PASSWORD }}
|
||||
|
||||
- name: Login to GitHub Container Registry
|
||||
if: env.HAS_GHCR_TOKEN == 'true'
|
||||
uses: docker/login-action@v4.2.0
|
||||
uses: docker/login-action@v4.4.0
|
||||
with:
|
||||
registry: ghcr.io
|
||||
username: ${{ env.GHCR_OWNER }}
|
||||
password: ${{ secrets.GHCR_GITHUB_TOKEN }}
|
||||
|
||||
- name: Build and Push Release Image
|
||||
uses: docker/build-push-action@v7.2.0
|
||||
uses: docker/build-push-action@v7.3.0
|
||||
with:
|
||||
context: .
|
||||
platforms: linux/amd64,linux/arm64
|
||||
|
||||
@@ -392,10 +392,16 @@ def _normalize_mcp_input_schema(schema: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
class MCPClient:
|
||||
def __init__(self) -> None:
|
||||
# Initialize session and client objects
|
||||
self.session: mcp.ClientSession | None = None
|
||||
self.exit_stack = AsyncExitStack()
|
||||
self._old_exit_stacks: list[AsyncExitStack] = [] # Track old stacks for cleanup
|
||||
|
||||
# Each connection runs in its own task so that anyio cancel scopes
|
||||
# are always exited from the task that entered them, preventing
|
||||
# RuntimeError: Attempted to exit cancel scope in a different task
|
||||
self._connection_task: asyncio.Task | None = None
|
||||
self._old_connection_tasks: list[asyncio.Task] = []
|
||||
|
||||
# Internal; managed exclusively by _run_connection.
|
||||
self.exit_stack: AsyncExitStack | None = None
|
||||
|
||||
self.name: str | None = None
|
||||
self.active: bool = True
|
||||
@@ -403,14 +409,67 @@ class MCPClient:
|
||||
self.server_errlogs: list[str] = []
|
||||
self.running_event = asyncio.Event()
|
||||
|
||||
# Store connection config for reconnection
|
||||
self._mcp_server_config: dict | None = None
|
||||
self._server_name: str | None = None
|
||||
self._reconnect_lock = asyncio.Lock() # Lock for thread-safe reconnection
|
||||
self._reconnecting: bool = False # For logging and debugging
|
||||
self._reconnecting: bool = False
|
||||
|
||||
async def _run_connection(
|
||||
self,
|
||||
mcp_server_config: dict,
|
||||
name: str,
|
||||
ready: asyncio.Future,
|
||||
) -> None:
|
||||
"""Own the full lifetime of one MCP connection.
|
||||
|
||||
This coroutine is always run inside a dedicated asyncio.Task
|
||||
(_connection_task). Because *this task* is the one that enters every
|
||||
anyio cancel scope (via sse_client / streamablehttp_client), anyio's
|
||||
_host_task check is always satisfied when the stack is later closed —
|
||||
either in the task's own finally block (normal path) or when the task
|
||||
is cancelled from outside (cleanup / reconnect path).
|
||||
|
||||
This avoids the
|
||||
RuntimeError: Attempted to exit cancel scope in a different task
|
||||
that previously occurred when aclose() was called from a different task
|
||||
or from the asyncio async-generator GC finalizer.
|
||||
"""
|
||||
# Capture the stack in a local variable so that if self.exit_stack is
|
||||
# overwritten by a concurrent _run_connection (during reconnect), this
|
||||
# task's finally block still closes only the resources it opened.
|
||||
stack = self.exit_stack = AsyncExitStack()
|
||||
try:
|
||||
try:
|
||||
await self._do_connect(mcp_server_config, name)
|
||||
except Exception as exc:
|
||||
if not ready.done():
|
||||
ready.set_exception(exc)
|
||||
raise
|
||||
else:
|
||||
if not ready.done():
|
||||
ready.set_result(None)
|
||||
# Hold the connection open until cancelled.
|
||||
await asyncio.Event().wait()
|
||||
finally:
|
||||
try:
|
||||
await stack.aclose()
|
||||
except Exception as e:
|
||||
logger.debug(f"Error closing exit stack for {name}: {e}")
|
||||
# Clear the instance reference only if it still points to this task's
|
||||
# stack; a concurrent reconnect may have already replaced it.
|
||||
if self.exit_stack is stack:
|
||||
self.exit_stack = None
|
||||
# Guard against the task exiting before ready was resolved.
|
||||
if not ready.done():
|
||||
ready.set_exception(RuntimeError("Connection task exited early"))
|
||||
|
||||
async def connect_to_server(self, mcp_server_config: dict, name: str) -> None:
|
||||
"""Connect to MCP server
|
||||
"""Connect to MCP server by spawning a dedicated owner task.
|
||||
|
||||
The owner task (_connection_task) holds the AsyncExitStack and all
|
||||
anyio cancel scopes for the lifetime of this connection. To disconnect,
|
||||
cancel _connection_task — the finally block in _run_connection will call
|
||||
aclose() from within the correct task context.
|
||||
|
||||
If `url` parameter exists:
|
||||
1. When transport is specified as `streamable_http`, use Streamable HTTP connection.
|
||||
@@ -421,10 +480,47 @@ class MCPClient:
|
||||
mcp_server_config (dict): Configuration for the MCP server. See https://modelcontextprotocol.io/quickstart/server
|
||||
|
||||
"""
|
||||
# Store config for reconnection
|
||||
self._mcp_server_config = mcp_server_config
|
||||
self._server_name = name
|
||||
|
||||
ready: asyncio.Future = asyncio.get_running_loop().create_future()
|
||||
|
||||
# Defensively cancel any existing connection task that was not cleaned
|
||||
# up before this call (e.g. if connect_to_server is called twice).
|
||||
if self._connection_task and not self._connection_task.done():
|
||||
self._cancel_connection_task(self._connection_task)
|
||||
self._connection_task = None
|
||||
|
||||
self._connection_task = asyncio.create_task(
|
||||
self._run_connection(mcp_server_config, name, ready),
|
||||
name=f"mcp-conn:{name}",
|
||||
)
|
||||
|
||||
try:
|
||||
await ready
|
||||
except asyncio.CancelledError:
|
||||
# Caller was cancelled while waiting — tear down the connection task.
|
||||
# cancel() is asynchronous; the task will not finish until the next
|
||||
# event-loop iteration, so we track it in _old_connection_tasks so
|
||||
# that cleanup() can await it later.
|
||||
if self._connection_task and not self._connection_task.done():
|
||||
self._cancel_connection_task(self._connection_task)
|
||||
self._connection_task = None
|
||||
raise
|
||||
except Exception:
|
||||
# _do_connect raised; the connection task's finally block may still
|
||||
# be running (e.g. awaiting stack.aclose()). Track it so that
|
||||
# cleanup() can await it, but do NOT cancel it — we want the
|
||||
# finally block to finish cleaning up resources naturally.
|
||||
if self._connection_task and not self._connection_task.done():
|
||||
self._old_connection_tasks.append(self._connection_task)
|
||||
self._connection_task = None
|
||||
raise
|
||||
|
||||
async def _do_connect(self, mcp_server_config: dict, name: str) -> None:
|
||||
"""Internal: perform the actual connection inside _run_connection's task."""
|
||||
# exit_stack is always set by _run_connection before _do_connect is called.
|
||||
assert self.exit_stack is not None
|
||||
cfg = _prepare_config(mcp_server_config.copy())
|
||||
|
||||
def logging_callback(
|
||||
@@ -563,16 +659,32 @@ class MCPClient:
|
||||
self.tools = response.tools
|
||||
return response
|
||||
|
||||
def _cancel_connection_task(self, task: asyncio.Task) -> None:
|
||||
"""Cancel a connection owner task and track it until it finishes."""
|
||||
# Prune already-finished tasks to avoid accumulating references over
|
||||
# many reconnections in a long-running process.
|
||||
self._old_connection_tasks = [
|
||||
t for t in self._old_connection_tasks if not t.done()
|
||||
]
|
||||
if task.done():
|
||||
return
|
||||
task.cancel()
|
||||
self._old_connection_tasks.append(task)
|
||||
|
||||
async def _reconnect(self) -> None:
|
||||
"""Reconnect to the MCP server using the stored configuration.
|
||||
|
||||
Cancels the current _connection_task (which owns the exit_stack and all
|
||||
anyio cancel scopes) and starts a fresh one. Because each connection
|
||||
task enters and exits its own anyio cancel scope, there is no
|
||||
cross-task cancel-scope violation and no GC finalizer surprise.
|
||||
|
||||
Uses asyncio.Lock to ensure thread-safe reconnection in concurrent environments.
|
||||
|
||||
Raises:
|
||||
Exception: raised when reconnection fails
|
||||
"""
|
||||
async with self._reconnect_lock:
|
||||
# Check if already reconnecting (useful for logging)
|
||||
if self._reconnecting:
|
||||
logger.debug(
|
||||
f"MCP Client {self._server_name} is already reconnecting, skipping"
|
||||
@@ -588,17 +700,16 @@ class MCPClient:
|
||||
f"Attempting to reconnect to MCP server {self._server_name}..."
|
||||
)
|
||||
|
||||
# Save old exit_stack for later cleanup (don't close it now to avoid cancel scope issues)
|
||||
if self.exit_stack:
|
||||
self._old_exit_stacks.append(self.exit_stack)
|
||||
|
||||
# Mark old session as invalid
|
||||
# Cancel the old connection task. Its finally block will call
|
||||
# exit_stack.aclose() from within the correct task context, so
|
||||
# anyio cancel scopes are exited cleanly without triggering the
|
||||
# GC-finalizer busy-spin bug.
|
||||
if self._connection_task and not self._connection_task.done():
|
||||
self._cancel_connection_task(self._connection_task)
|
||||
self._connection_task = None
|
||||
self.session = None
|
||||
|
||||
# Create new exit stack for new connection
|
||||
self.exit_stack = AsyncExitStack()
|
||||
|
||||
# Reconnect using stored config
|
||||
# Reconnect — this creates a new _connection_task.
|
||||
await self.connect_to_server(self._mcp_server_config, self._server_name)
|
||||
await self.list_tools_and_save()
|
||||
|
||||
@@ -663,19 +774,20 @@ class MCPClient:
|
||||
return await _call_with_retry()
|
||||
|
||||
async def cleanup(self) -> None:
|
||||
"""Clean up resources including old exit stacks from reconnections"""
|
||||
# Close current exit stack
|
||||
try:
|
||||
await self.exit_stack.aclose()
|
||||
except Exception as e:
|
||||
logger.debug(f"Error closing current exit stack: {e}")
|
||||
"""Clean up resources by cancelling the connection owner task."""
|
||||
# Cancel current and any old connection tasks via the shared helper so
|
||||
# all cancellation + tracking behaviour goes through one code path.
|
||||
if self._connection_task:
|
||||
self._cancel_connection_task(self._connection_task)
|
||||
self._connection_task = None
|
||||
|
||||
# Don't close old exit stacks as they may be in different task contexts
|
||||
# They will be garbage collected naturally
|
||||
# Just clear the list to release references
|
||||
self._old_exit_stacks.clear()
|
||||
if self._old_connection_tasks:
|
||||
pending = [t for t in self._old_connection_tasks if not t.done()]
|
||||
if pending:
|
||||
await asyncio.gather(*pending, return_exceptions=True)
|
||||
self._old_connection_tasks.clear()
|
||||
|
||||
# Set running_event first to unblock any waiting tasks
|
||||
# Set running_event to unblock any waiting tasks
|
||||
self.running_event.set()
|
||||
|
||||
|
||||
|
||||
@@ -502,6 +502,7 @@ class QQOfficialMessageEvent(AstrMessageEvent):
|
||||
logger.info("[QQOfficial] 回复消息失败: %s, 尝试使用主动发送接口。", err)
|
||||
if payload.get("msg_id"):
|
||||
fallback_payload = payload.copy()
|
||||
fallback_payload.pop("msg_id", None)
|
||||
try:
|
||||
ret = await send_func(fallback_payload)
|
||||
logger.info("[QQOfficial] 使用主动发送接口发送成功。")
|
||||
|
||||
@@ -723,8 +723,21 @@ class FunctionToolManager:
|
||||
logger.debug(f"MCP client {name} task was cancelled")
|
||||
raise
|
||||
finally:
|
||||
# Cleanup in the same task that entered the anyio contexts
|
||||
await asyncio.shield(self._terminate_mcp_client(name))
|
||||
# Cleanup in the same task that entered the anyio contexts:
|
||||
# asyncio.shield() would schedule the coroutine as a separate
|
||||
# Task, and anyio cancel scopes cannot exit across tasks (#9068).
|
||||
# Absorb late cancellations so a forced shutdown cannot abort
|
||||
# the cleanup halfway.
|
||||
task = asyncio.current_task()
|
||||
while True:
|
||||
try:
|
||||
await self._terminate_mcp_client(name)
|
||||
break
|
||||
except asyncio.CancelledError:
|
||||
# Task.uncancel() is 3.11+; on 3.10 absorbing the
|
||||
# cancellation is sufficient.
|
||||
if task is not None and hasattr(task, "uncancel"):
|
||||
task.uncancel()
|
||||
|
||||
lifecycle_task = asyncio.create_task(
|
||||
connect_and_lifecycle(), name=f"mcp-client:{name}"
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
import pytest
|
||||
@@ -348,6 +349,65 @@ def test_firecrawl_tools_are_registered_as_builtin_tools():
|
||||
assert manager.is_builtin_tool("firecrawl_extract_web_page") is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mcp_shutdown_cleanup_runs_in_lifecycle_task(monkeypatch):
|
||||
"""Disabling an MCP server must clean up in the task that connected.
|
||||
|
||||
anyio cancel scopes entered in connect_to_server() can only be exited
|
||||
from the same task, otherwise the scope state is corrupted and its
|
||||
cancellation loop spins at 100% CPU (#9068).
|
||||
"""
|
||||
manager = FunctionToolManager()
|
||||
seen = {}
|
||||
|
||||
async def fake_connect(self, config, name):
|
||||
seen["connect_task"] = asyncio.current_task()
|
||||
|
||||
async def fake_list_tools(self):
|
||||
self.tools = []
|
||||
|
||||
async def fake_cleanup(self):
|
||||
seen["cleanup_task"] = asyncio.current_task()
|
||||
|
||||
monkeypatch.setattr(ftm.MCPClient, "connect_to_server", fake_connect)
|
||||
monkeypatch.setattr(ftm.MCPClient, "list_tools_and_save", fake_list_tools)
|
||||
monkeypatch.setattr(ftm.MCPClient, "cleanup", fake_cleanup)
|
||||
|
||||
await manager.enable_mcp_server("dummy", {"command": "python"}, timeout=5)
|
||||
await manager.disable_mcp_server("dummy", timeout=5)
|
||||
|
||||
assert seen["cleanup_task"] is seen["connect_task"]
|
||||
assert "dummy" not in manager.mcp_client_dict
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mcp_shutdown_cleanup_survives_late_cancellation(monkeypatch):
|
||||
"""A cancellation arriving mid-cleanup must not abort the cleanup."""
|
||||
manager = FunctionToolManager()
|
||||
cleanup_calls = []
|
||||
|
||||
async def fake_connect(self, config, name):
|
||||
pass
|
||||
|
||||
async def fake_list_tools(self):
|
||||
self.tools = []
|
||||
|
||||
async def fake_cleanup(self):
|
||||
cleanup_calls.append(asyncio.current_task())
|
||||
if len(cleanup_calls) == 1:
|
||||
raise asyncio.CancelledError()
|
||||
|
||||
monkeypatch.setattr(ftm.MCPClient, "connect_to_server", fake_connect)
|
||||
monkeypatch.setattr(ftm.MCPClient, "list_tools_and_save", fake_list_tools)
|
||||
monkeypatch.setattr(ftm.MCPClient, "cleanup", fake_cleanup)
|
||||
|
||||
await manager.enable_mcp_server("dummy", {"command": "python"}, timeout=5)
|
||||
await manager.disable_mcp_server("dummy", timeout=5)
|
||||
|
||||
assert len(cleanup_calls) == 2
|
||||
assert "dummy" not in manager.mcp_client_dict
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_modelscope_sync_enables_only_synced_servers(monkeypatch):
|
||||
class FakeResponse:
|
||||
|
||||
Reference in New Issue
Block a user