diff --git a/core/main_terminal_parts/tools_definition.py b/core/main_terminal_parts/tools_definition.py index a2bc422..ffff9d9 100644 --- a/core/main_terminal_parts/tools_definition.py +++ b/core/main_terminal_parts/tools_definition.py @@ -126,6 +126,10 @@ class MainTerminalToolsDefinitionMixin: } for tool in tools: func = tool.get("function") or {} + tool_name = str(func.get("name") or "").strip() + # MCP 扩展工具参数需与远端 schema 严格一致，不注入 intent + if tool_name.startswith("mcp__"): + continue params = func.get("parameters") or {} if not isinstance(params, dict): continue diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index b410f80..e252cd5 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -695,10 +695,15 @@ class MainTerminalToolsExecutionMixin: if manager is None: result = {"success": False, "error": "MCP 管理器不可用"} else: + # intent 仅用于原生工具的人类可读展示，不应透传给 MCP 远端工具 + mcp_arguments = arguments + if isinstance(arguments, dict) and "intent" in arguments: + mcp_arguments = dict(arguments) + mcp_arguments.pop("intent", None) result = await asyncio.to_thread( manager.call_tool_by_alias, tool_name, - arguments, + mcp_arguments, alias_map=getattr(self, "mcp_tool_alias_map", None), ) elif custom_tool: diff --git a/test/test_mcp_integration.py b/test/test_mcp_integration.py index 3a07c4d..baad8b0 100644 --- a/test/test_mcp_integration.py +++ b/test/test_mcp_integration.py @@ -118,6 +118,179 @@ class MCPIntegrationTest(unittest.TestCase): 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),