这个mcp两个client同时连接第2个会阻塞

[gauss-clb]

大致可以参考下面代码，这里的Agent调用了qwen-agent，agent_load = Agent(enable_thinking=True, verbose=True)这行代码是为了加载MCPManager，因为qwen-agent这个框架里这个类是单例的，直接并发加载会报错，先提前加载一下。

import asyncio
from agent_cls import Agent

agent_load = Agent(enable_thinking=True, verbose=True)

def call():
    import time
    st = time.time()
    agent = Agent(enable_thinking=True, verbose=True)
    agent.set_system_message('hello')
    ed = time.time()
    print(f'Cost: {ed-st}', flush=True)


async def init_agent(name):
    print(f"[{name}] init ...")
    await asyncio.to_thread(call)

async def main():
    # 并发启动两个 agent
    print("Starting two agents in parallel...\n")

    await asyncio.gather(
        init_agent("Agent-1"),
        init_agent("Agent-2"),
    )

if __name__ == "__main__":
    asyncio.run(main())

这个是qwen-agent加载mcp的代码部分，箭头指的部分只会打印1次，另1个就卡在这里，等300s超时。

[hustcc]

@gauss-clb 可以在你的环境帮忙调试解决一下试试吗？

[gauss-clb]

@gauss-clb 可以在你的环境帮忙调试解决一下试试吗？

你可以提供代码，我帮你在我的环境里运行。我换了另一个mcp，是正常的。但是mcp-server-chart就会卡住。主要我看不懂JS代码。

[gauss-clb]

@hustcc 看这个代码吧，这个不需要qwen-agent了，我剥离出来了，只需要改一下ip就行，第1个成功，第2个会卡在await self.session.initialize()

from contextlib import AsyncExitStack
from typing import Dict, Optional, Union
from datetime import datetime
import threading
import asyncio

class MCPClient:

    def __init__(self):
        from mcp import ClientSession
        self.session: Optional[ClientSession] = None
        self.tools: list = None
        self.exit_stack = AsyncExitStack()
        self.resources: bool = False
        self._last_mcp_server_name = None
        self._last_mcp_server = None
        self.client_id = None  # For replacing in MCPManager.clients

    async def connection_server(self, mcp_server_name, mcp_server):
        from mcp import ClientSession, StdioServerParameters
        from mcp.client.sse import sse_client
        from mcp.client.stdio import stdio_client
        from mcp.client.streamable_http import streamablehttp_client
        """Connect to an MCP server and retrieve the available tools."""
        # Save parameters
        self._last_mcp_server_name = mcp_server_name
        self._last_mcp_server = mcp_server

        try:
            if 'url' in mcp_server:
                url = mcp_server.get('url')
                sse_read_timeout = mcp_server.get('sse_read_timeout', 300)
                print(f'{mcp_server_name} sse_read_timeout: {sse_read_timeout}s')
                if mcp_server.get('type', 'sse') == 'streamable-http':
                    # streamable-http mode
                    """streamable-http mode mcp example:
                    {"mcpServers": {
                            "streamable-mcp-server": {
                            "type": "streamable-http",
                            "url":"http://0.0.0.0:8000/mcp"
                            }
                        }
                    }
                    """
                    self._streams_context = streamablehttp_client(
                        url=url, sse_read_timeout=datetime.timedelta(seconds=sse_read_timeout))
                    read_stream, write_stream, get_session_id = await self.exit_stack.enter_async_context(
                        self._streams_context)
                    self._session_context = ClientSession(read_stream, write_stream)
                    self.session = await self.exit_stack.enter_async_context(self._session_context)
                else:
                    # sse mode
                    headers = mcp_server.get('headers', {'Accept': 'text/event-stream'})
                    self._streams_context = sse_client(url, headers, sse_read_timeout=sse_read_timeout)
                    streams = await self.exit_stack.enter_async_context(self._streams_context)
                    self._session_context = ClientSession(*streams)
                    self.session = await self.exit_stack.enter_async_context(self._session_context)
            else:
                server_params = StdioServerParameters(command=mcp_server['command'],
                                                      args=mcp_server['args'],
                                                      env=mcp_server.get('env', None))
                stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
                self.stdio, self.write = stdio_transport
                self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
                print(
                    f'Initializing a MCP stdio_client, if this takes forever, please check the config of this mcp server: {mcp_server_name}'
                )

            print('before session initialize')
            await self.session.initialize()
            print('after session initialize')
            list_tools = await self.session.list_tools()
            self.tools = list_tools.tools
            # print('after enter_async_context')
            try:
                list_resources = await self.session.list_resources()  # Check if the server has resources
                if list_resources.resources:
                    self.resources = True
            except Exception:
                # logger.info(f"No list resources: {e}")
                pass
        except Exception as e:
            print(f'Failed in connecting to MCP server: {e}')
            raise e


class MCPManager:
    _instance = None  # Private class variable to store the unique instance

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(MCPManager, cls).__new__(cls, *args, **kwargs)
        return cls._instance

    def __init__(self):
        if not hasattr(self, 'clients'):  # The singleton should only be inited once
            """Set a new event loop in a separate thread"""
            try:
                import mcp  # noqa
            except ImportError as e:
                raise ImportError('Could not import mcp. Please install mcp with `pip install -U mcp`.') from e

            # load_dotenv()  # Load environment variables from .env file
            self.clients: dict = {}
            self.loop = asyncio.new_event_loop()
            self.loop_thread = threading.Thread(target=self.start_loop, daemon=True)
            self.loop_thread.start()

            # A fallback way to terminate MCP tool processes after Qwen-Agent exits
            self.processes = []

    def initConfig(self, config: Dict):
        print(f'Initializing MCP tools from mcp servers: {list(config["mcpServers"].keys())}')
        # Submit coroutine to the event loop and wait for the result
        future = asyncio.run_coroutine_threadsafe(self.init_config_async(config), self.loop)
        try:
            result = future.result()  # You can specify a timeout if desired
            return result
        except Exception as e:
            print(f'Failed in initializing MCP tools: {e}')
            raise e

    async def init_config_async(self, config: Dict):
        mcp_servers = config['mcpServers']
        for server_name in mcp_servers:
            client = MCPClient()
            server = mcp_servers[server_name]
            await client.connection_server(mcp_server_name=server_name,
                                           mcp_server=server)  # Attempt to connect to the server
            print(f'{server_name}: finish')


    def start_loop(self):
        asyncio.set_event_loop(self.loop)

        # Set a global exception handler to silently handle cross-task exceptions from MCP SSE connections
        def exception_handler(loop, context):
            exception = context.get('exception')
            if exception:
                # Silently handle cross-task exceptions from MCP SSE connections
                if (isinstance(exception, RuntimeError) and
                        'Attempted to exit cancel scope in a different task' in str(exception)):
                    return  # Silently ignore this type of exception
                if (isinstance(exception, BaseExceptionGroup) and  # noqa
                        'Attempted to exit cancel scope in a different task' in str(exception)):  # noqa
                    return  # Silently ignore this type of exception

            # Other exceptions are handled normally
            loop.default_exception_handler(context)

        self.loop.set_exception_handler(exception_handler)
        self.loop.run_forever()

manager = MCPManager()
ip = "xx.xx.xxx.xxx"
config = {
    "mcpServers": {
        "mcp-server-chart": {
            "url": f"http://{ip}:1122/sse"
        }
    }
}

def call():
    import time
    st = time.time()
    manager.initConfig(config)
    ed = time.time()
    print(f'Cost: {ed-st}', flush=True)


async def init_agent(name):
    print(f"[{name}] init ...")
    await asyncio.to_thread(call)

async def main():
    # 并发启动两个 agent
    print("Starting two agents in parallel...\n")

    await asyncio.gather(
        init_agent("Agent-1"),
        init_agent("Agent-2"),
    )

if __name__ == "__main__":
    asyncio.run(main())

[gauss-clb]

@hustcc 以下gpt5分析，应该是mcp-server-chart只支持1个client连接？

[gauss-clb]

@hustcc docker相关配置

# Use Node.js 18 as the base image
FROM swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/python:3.11

# Set workdir
WORKDIR /app

RUN apt update
RUN apt install -y nodejs npm
# Global install of @antv/mcp-server-chart
RUN npm install -g @antv/mcp-server-chart@0.9.0-beta.0

# Start the server (using streamable for transmission)
CMD ["mcp-server-chart", "--transport", "sse", "--port", "1122"]

  mcp-server-chart:
    build: ./mcp-server-chart/sse
    ports:
      - 1122:1122
    volumes:
      - ./mcp-server-chart/sse:/mcp-server-chart/sse
    environment:
      - NODE_ENV=production
      - TZ=Asia/Shanghai
      - DISABLED_TOOLS=generate_district_map,generate_path_map,generate_pin_map
    restart: unless-stopped

[hustcc]

感谢反馈，我这周用你的代码调试看看。

[hustcc]

@gauss-clb #233

你本地试试这个分支的代码？我在单测中用 JavaScript 的 mcp client 复现出来了，现在单元测试的通过的。

[gauss-clb]

@gauss-clb #233

你本地试试这个分支的代码？我在单测中用 JavaScript 的 mcp client 复现出来了，现在单元测试的通过的。

还是卡住了

[hustcc]

你是怎么本地启动的？ @gauss-clb

[gauss-clb]

你是怎么本地启动的？ @gauss-clb

1. npm install
2. npm run build
3. node build/index.js -t sse

[gauss-clb]

@hustcc 你测试是成功的嘛？

[hustcc]

@gauss-clb 成功的，而且还加了单元测试了。

[gauss-clb]

@gauss-clb 成功的，而且还加了单元测试了。

能贴下你的测试代码嘛？