from __future__ import annotations import asyncio import json import sys import tempfile import textwrap import unittest from pathlib import Path from types import SimpleNamespace from core.main_terminal import MainTerminal from modules.mcp_client_manager import MCPClientManager, _StdioMCPClient from modules.mcp_server_registry import MCPServerRegistry class MCPIntegrationTest(unittest.TestCase): def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() root = Path(self.temp_dir.name) self.project_dir = root / "project" self.project_dir.mkdir(parents=True, exist_ok=True) self.data_dir = root / "data" self.data_dir.mkdir(parents=True, exist_ok=True) self.registry_path = self.data_dir / "mcp_servers.json" self.registry = MCPServerRegistry(config_path=str(self.registry_path), enabled=True) self.manager = MCPClientManager(self.registry) demo_script = Path(__file__).resolve().parents[1] / "scripts" / "mcp_demo_server.py" self.assertTrue(demo_script.exists(), f"missing demo script: {demo_script}") self.demo_script = demo_script self.registry.upsert_server( { "id": "demo_stdio", "name": "Demo MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(self.demo_script)], "timeout_seconds": 10, } ) def _write_temp_server(self, filename: str, content: str) -> Path: path = self.project_dir / filename path.write_text(textwrap.dedent(content).strip() + "\n", encoding="utf-8") return path def tearDown(self): try: self.manager.close_all_clients() except Exception: pass self.temp_dir.cleanup() def test_manager_discover_and_call(self): sync_result = self.manager.sync_servers(server_id="demo_stdio") self.assertTrue(sync_result.get("success"), sync_result) self.assertGreaterEqual(sync_result.get("tools_count", 0), 2, sync_result) tools, alias_map = self.manager.build_openai_tools(ensure_discovery=False) self.assertTrue(tools) echo_alias = None for alias, binding in alias_map.items(): if binding.remote_name == "echo_upper": echo_alias = alias break self.assertTrue(echo_alias, f"missing echo_upper alias: {list(alias_map.keys())}") call_result = self.manager.call_tool_by_alias( echo_alias, {"text": "hello mcp"}, alias_map=alias_map, ) self.assertTrue(call_result.get("success"), call_result) payload = json.dumps(call_result, ensure_ascii=False) self.assertIn("HELLO MCP", payload) def test_main_terminal_tool_execution(self): sync_result = self.manager.sync_servers(server_id="demo_stdio") self.assertTrue(sync_result.get("success"), sync_result) terminal = MainTerminal( project_path=str(self.project_dir), thinking_mode=False, data_dir=str(self.data_dir), ) terminal.mcp_tools_enabled = True terminal.mcp_server_registry = self.registry terminal.mcp_client_manager = self.manager tool_defs = terminal.define_tools() alias = None for item in tool_defs: name = ((item or {}).get("function") or {}).get("name") if isinstance(name, str) and name.startswith("mcp__") and "echo_upper" in name: alias = name break if not alias: for item in tool_defs: name = ((item or {}).get("function") or {}).get("name") if isinstance(name, str) and name.startswith("mcp__"): alias = name break self.assertTrue(alias, "no mcp alias in define_tools") raw = asyncio.run(terminal.handle_tool_call(alias, {"text": "abc"})) data = json.loads(raw) self.assertTrue(data.get("success"), data) self.assertIn("ABC", json.dumps(data, ensure_ascii=False)) # 原生 MCP 列表工具 list_raw = asyncio.run(terminal.handle_tool_call("list_mcp_servers", {"refresh": False})) list_data = json.loads(list_raw) self.assertTrue(list_data.get("success"), list_data) self.assertGreaterEqual(int(list_data.get("count", 0)), 1, list_data) def test_main_terminal_mcp_call_drops_intent_argument(self): strict_script = self._write_temp_server( "strict_args_mcp_server.py", """ import json import sys def send(payload): sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\\n") sys.stdout.flush() while True: line = sys.stdin.readline() if not line: break line = line.strip() if not line: continue req = json.loads(line) method = req.get("method") msg_id = req.get("id") params = req.get("params") or {} if msg_id is None: continue if method == "initialize": send({ "jsonrpc": "2.0", "id": msg_id, "result": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {}}, "serverInfo": {"name": "strict-args", "version": "1.0.0"}, }, }) continue if method == "tools/list": send({ "jsonrpc": "2.0", "id": msg_id, "result": { "tools": [ { "name": "echo_strict", "description": "仅接受 text 参数", "inputSchema": { "type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"], }, } ] }, }) continue if method == "tools/call": name = str(params.get("name") or "") arguments = params.get("arguments") if isinstance(params.get("arguments"), dict) else {} keys = sorted(arguments.keys()) if name != "echo_strict": send({ "jsonrpc": "2.0", "id": msg_id, "result": { "isError": True, "content": [{"type": "text", "text": "unknown tool"}], }, }) continue if keys != ["text"]: send({ "jsonrpc": "2.0", "id": msg_id, "result": { "isError": True, "content": [{"type": "text", "text": "unexpected keys:" + ",".join(keys)}], "structuredContent": {"received_keys": keys}, }, }) continue text = str(arguments.get("text") or "") send({ "jsonrpc": "2.0", "id": msg_id, "result": { "isError": False, "content": [{"type": "text", "text": text.upper()}], "structuredContent": {"received_keys": keys, "text": text.upper()}, }, }) continue send({ "jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Method not found: {method}"}, }) """, ) self.registry.upsert_server( { "id": "strict_stdio", "name": "Strict MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(strict_script)], "timeout_seconds": 10, } ) sync_result = self.manager.sync_servers(server_id="strict_stdio") self.assertTrue(sync_result.get("success"), sync_result) terminal = MainTerminal( project_path=str(self.project_dir), thinking_mode=False, data_dir=str(self.data_dir), ) terminal.mcp_tools_enabled = True terminal.mcp_server_registry = self.registry terminal.mcp_client_manager = self.manager alias = None for item in terminal.define_tools(): name = ((item or {}).get("function") or {}).get("name") if isinstance(name, str) and name.startswith("mcp__strict_stdio__"): alias = name break self.assertTrue(alias, "missing strict mcp alias") raw = asyncio.run(terminal.handle_tool_call(alias, {"intent": "测试调用", "text": "abc"})) data = json.loads(raw) self.assertTrue(data.get("success"), data) self.assertIn("ABC", json.dumps(data, ensure_ascii=False)) self.assertEqual((data.get("structured_content") or {}).get("received_keys"), ["text"], data) def test_define_tools_skip_intent_in_mcp_alias(self): sync_result = self.manager.sync_servers(server_id="demo_stdio") self.assertTrue(sync_result.get("success"), sync_result) terminal = MainTerminal( project_path=str(self.project_dir), thinking_mode=False, data_dir=str(self.data_dir), ) terminal.mcp_tools_enabled = True terminal.mcp_server_registry = self.registry terminal.mcp_client_manager = self.manager terminal.tool_intent_enabled = True tool_defs = terminal.define_tools() mcp_tool = None native_tool = None for item in tool_defs: fn = (item or {}).get("function") or {} name = str(fn.get("name") or "") if not native_tool and name == "run_command": native_tool = item if not mcp_tool and name.startswith("mcp__demo_stdio__"): mcp_tool = item if mcp_tool and native_tool: break self.assertIsNotNone(native_tool, "missing native run_command tool") self.assertIsNotNone(mcp_tool, "missing mcp alias tool") native_params = ((native_tool or {}).get("function") or {}).get("parameters") or {} native_props = native_params.get("properties") if isinstance(native_params, dict) else {} self.assertIn("intent", native_props or {}) mcp_params = ((mcp_tool or {}).get("function") or {}).get("parameters") or {} mcp_props = mcp_params.get("properties") if isinstance(mcp_params, dict) else {} self.assertNotIn("intent", mcp_props or {}) self.assertNotIn("intent", mcp_params.get("required") or []) def test_main_terminal_mcp_blocked_in_docker_mode(self): terminal = MainTerminal( project_path=str(self.project_dir), thinking_mode=False, data_dir=str(self.data_dir), ) terminal.mcp_tools_enabled = True terminal.mcp_server_registry = self.registry terminal.mcp_client_manager = self.manager terminal.container_session = SimpleNamespace(mode="docker") message = "当前为docker模式，MCP仅支持宿主机模式" list_raw = asyncio.run(terminal.handle_tool_call("list_mcp_servers", {"refresh": True})) list_data = json.loads(list_raw) self.assertFalse(list_data.get("success"), list_data) self.assertEqual(list_data.get("error"), message, list_data) call_raw = asyncio.run(terminal.handle_tool_call("mcp__demo_stdio__echo_upper", {"text": "abc"})) call_data = json.loads(call_raw) self.assertFalse(call_data.get("success"), call_data) self.assertEqual(call_data.get("error"), message, call_data) def test_calculator_mcp_server(self): calculator_script = Path(__file__).resolve().parents[1] / "scripts" / "mcp_calculator_server.py" self.assertTrue(calculator_script.exists(), f"missing calculator script: {calculator_script}") self.registry.upsert_server( { "id": "calc_stdio", "name": "Calculator MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(calculator_script)], "timeout_seconds": 10, } ) sync_result = self.manager.sync_servers(server_id="calc_stdio") self.assertTrue(sync_result.get("success"), sync_result) self.assertEqual(sync_result.get("tools_count"), 1, sync_result) _, alias_map = self.manager.build_openai_tools(ensure_discovery=False) calc_alias = None for alias, binding in alias_map.items(): if binding.server_id == "calc_stdio" and binding.remote_name == "calculator": calc_alias = alias break self.assertTrue(calc_alias, f"missing calculator alias: {list(alias_map.keys())}") call_result = self.manager.call_tool_by_alias( calc_alias, {"operation": "mul", "a": 6, "b": 7}, alias_map=alias_map, ) self.assertTrue(call_result.get("success"), call_result) self.assertIn("42", json.dumps(call_result, ensure_ascii=False)) def test_sync_servers_reload_external_file_changes(self): # 模拟“外部工具直接写 mcp_servers.json”，绕过 registry 内存缓存 payload = { "servers": [ { "id": "demo_stdio", "name": "Demo MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(self.demo_script)], "timeout_seconds": 10, }, { "id": "extra_stdio", "name": "Extra MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(self.demo_script)], "timeout_seconds": 10, }, ] } self.registry_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") sync_result = self.manager.sync_servers() self.assertTrue(sync_result.get("success"), sync_result) self.assertEqual(sync_result.get("servers_total"), 2, sync_result) by_id = {item.get("server_id"): item for item in sync_result.get("results") or []} self.assertIn("extra_stdio", by_id, sync_result) self.assertTrue(by_id["extra_stdio"].get("success"), sync_result) def test_build_openai_tools_reload_external_file_changes(self): # 模拟外部直接改配置文件（registry._cache 仍是旧内容） payload = { "servers": [ { "id": "demo_stdio", "name": "Demo MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(self.demo_script)], "timeout_seconds": 10, }, { "id": "extra_stdio", "name": "Extra MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(self.demo_script)], "timeout_seconds": 10, }, ] } self.registry_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") tools, alias_map = self.manager.build_openai_tools( ensure_discovery=True, discovery_timeout_seconds=10, ) self.assertTrue(tools) self.assertTrue(alias_map) self.assertTrue( any(getattr(binding, "server_id", "") == "extra_stdio" for binding in alias_map.values()), alias_map, ) # 关键断言：外部新增服务不应被旧缓存反写覆盖 after = json.loads(self.registry_path.read_text(encoding="utf-8")) ids = {str((item or {}).get("id") or "") for item in (after.get("servers") or [])} self.assertIn("demo_stdio", ids) self.assertIn("extra_stdio", ids) def test_stdio_container_path_rewrite_from_workspace(self): fake_session = SimpleNamespace( mode="docker", container_name="fake-container", mount_path="/workspace", workspace_path=str(self.project_dir), sandbox_bin="docker", ) server = { "command": "python3", "args": [ str(self.project_dir / "scripts" / "demo.py"), f"--root={self.project_dir / 'subdir'}", ], "cwd": str(self.project_dir), "env": {"HELLO": "1"}, } client = _StdioMCPClient(server, timeout_seconds=10, protocol_version="2025-06-18", container_session=fake_session) launch_cmd, cwd, _ = client._prepare_launch() self.assertTrue(str(launch_cmd[0]).endswith("docker"), launch_cmd) self.assertEqual(launch_cmd[1:3], ["exec", "-i"], launch_cmd) self.assertIn("-w", launch_cmd) self.assertEqual(cwd, None) joined = " ".join(launch_cmd) self.assertIn("/workspace/scripts/demo.py", joined) self.assertIn("--root=/workspace/subdir", joined) self.assertIn("-e HELLO=1", joined) def test_stdio_container_command_fallback_for_host_abs_path(self): fake_session = SimpleNamespace( mode="docker", container_name="fake-container", mount_path="/workspace", workspace_path=str(self.project_dir), sandbox_bin="docker", ) server = { "command": "/opt/homebrew/bin/npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/jojo/Desktop"], "cwd": "", "env": {}, } client = _StdioMCPClient(server, timeout_seconds=10, protocol_version="2025-06-18", container_session=fake_session) launch_cmd, cwd, _ = client._prepare_launch() self.assertEqual(cwd, None) self.assertIn("fake-container", launch_cmd) self.assertIn("npx", launch_cmd) self.assertNotIn("/opt/homebrew/bin/npx", launch_cmd) self.assertIn("/Users/jojo/Desktop", launch_cmd) def test_stateful_stdio_server_reuses_persistent_session(self): stateful_script = self._write_temp_server( "stateful_mcp_server.py", """ import json import sys counter = 0 def send(payload): sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\\n") sys.stdout.flush() while True: line = sys.stdin.readline() if not line: break line = line.strip() if not line: continue req = json.loads(line) method = req.get("method") msg_id = req.get("id") params = req.get("params") or {} if msg_id is None: continue if method == "initialize": send({ "jsonrpc": "2.0", "id": msg_id, "result": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {}}, "serverInfo": {"name": "stateful", "version": "1.0.0"}, }, }) continue if method == "tools/list": send({ "jsonrpc": "2.0", "id": msg_id, "result": { "tools": [ { "name": "next_counter", "description": "返回递增计数器", "inputSchema": {"type": "object", "properties": {}}, } ] }, }) continue if method == "tools/call": name = str(params.get("name") or "") if name == "next_counter": counter += 1 send({ "jsonrpc": "2.0", "id": msg_id, "result": { "isError": False, "content": [{"type": "text", "text": str(counter)}], "structuredContent": {"counter": counter}, }, }) continue send({ "jsonrpc": "2.0", "id": msg_id, "result": { "isError": True, "content": [{"type": "text", "text": "unknown tool"}], }, }) continue send({ "jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Method not found: {method}"}, }) """, ) self.registry.upsert_server( { "id": "stateful_stdio", "name": "Stateful MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(stateful_script)], "timeout_seconds": 10, } ) sync_result = self.manager.sync_servers(server_id="stateful_stdio") self.assertTrue(sync_result.get("success"), sync_result) _, alias_map = self.manager.build_openai_tools(ensure_discovery=False) alias = None for key, binding in alias_map.items(): if binding.server_id == "stateful_stdio" and binding.remote_name == "next_counter": alias = key break self.assertTrue(alias, f"missing alias for stateful server: {list(alias_map.keys())}") first = self.manager.call_tool_by_alias(alias, {}, alias_map=alias_map) second = self.manager.call_tool_by_alias(alias, {}, alias_map=alias_map) self.assertTrue(first.get("success"), first) self.assertTrue(second.get("success"), second) self.assertEqual((first.get("structured_content") or {}).get("counter"), 1, first) self.assertEqual((second.get("structured_content") or {}).get("counter"), 2, second) def test_stdio_server_request_ping_is_handled(self): ping_script = self._write_temp_server( "ping_request_mcp_server.py", """ import json import select import sys import time def send(payload): sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\\n") sys.stdout.flush() def read_json_line(timeout_seconds): deadline = time.monotonic() + timeout_seconds while time.monotonic() < deadline: remain = max(0.05, deadline - time.monotonic()) readable, _, _ = select.select([sys.stdin], [], [], min(0.2, remain)) if not readable: continue line = sys.stdin.readline() if not line: return None line = line.strip() if not line: continue try: return json.loads(line) except Exception: continue return None while True: line = sys.stdin.readline() if not line: break line = line.strip() if not line: continue req = json.loads(line) method = req.get("method") msg_id = req.get("id") params = req.get("params") or {} if msg_id is None: continue if method == "initialize": send({ "jsonrpc": "2.0", "id": msg_id, "result": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {}}, "serverInfo": {"name": "ping-stdio", "version": "1.0.0"}, }, }) continue if method == "tools/list": send({ "jsonrpc": "2.0", "id": msg_id, "result": { "tools": [ { "name": "trigger_ping", "description": "触发服务端 ping 请求", "inputSchema": {"type": "object", "properties": {}}, } ] }, }) continue if method == "tools/call": ping_id = 9001 send({ "jsonrpc": "2.0", "id": ping_id, "method": "ping", "params": {}, }) response = read_json_line(2.5) or {} ok = ( isinstance(response, dict) and response.get("id") == ping_id and isinstance(response.get("result"), dict) ) send({ "jsonrpc": "2.0", "id": msg_id, "result": { "isError": not ok, "content": [{"type": "text", "text": "ping_ok" if ok else "ping_failed"}], "structuredContent": {"ping_ok": ok}, }, }) continue send({ "jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"Method not found: {method}"}, }) """, ) self.registry.upsert_server( { "id": "ping_stdio", "name": "Ping MCP", "transport": "stdio", "enabled": True, "command": sys.executable, "args": [str(ping_script)], "timeout_seconds": 10, } ) sync_result = self.manager.sync_servers(server_id="ping_stdio") self.assertTrue(sync_result.get("success"), sync_result) _, alias_map = self.manager.build_openai_tools(ensure_discovery=False) alias = None for key, binding in alias_map.items(): if binding.server_id == "ping_stdio" and binding.remote_name == "trigger_ping": alias = key break self.assertTrue(alias, f"missing alias for ping server: {list(alias_map.keys())}") result = self.manager.call_tool_by_alias(alias, {}, alias_map=alias_map) self.assertTrue(result.get("success"), result) self.assertTrue((result.get("structured_content") or {}).get("ping_ok"), result) def test_build_call_message_empty_content_returns_empty_string(self): message = MCPClientManager._build_call_message( { "content": [ { "type": "text", "text": "", } ] } ) self.assertEqual(message, "") if __name__ == "__main__": unittest.main()