"""Deployed MCP tools must reach the model and be dispatched by name.""" import json from threading import Event import pytest from mulit_agent_app.application.task_executor import TaskExecutor from mulit_agent_app.infrastructure.database import AgentDatabase from mulit_agent_app.infrastructure.mcp_client import ( HttpTransport, McpClient, McpClientError, open_client, ) from mulit_agent_app.infrastructure.openai_compatible_provider import OpenAICompatibleProvider from mulit_agent_app.infrastructure.tool_registry import McpEndpoint, Toolbox _REMOTE_TOOLS = [ { "name": "dom_query", "description": "查询页面 DOM。", "inputSchema": {"type": "object", "properties": {"selector": {"type": "string"}}, "required": ["selector"]}, } ] class _Credentials: def load_llm_api_key(self): return "test-key" class _FakeMcpClient: """Stands in for a live MCP connection and records every call.""" def __init__(self, calls: list[tuple[str, dict]]) -> None: self.calls = calls self.closed = False def call_tool(self, name: str, arguments: dict) -> str: self.calls.append((name, arguments)) return "MCP 工具结果" def close(self) -> None: self.closed = True def _endpoint(calls: list[tuple[str, dict]]) -> McpEndpoint: return McpEndpoint("imageforge", "test-definition", _FakeMcpClient(calls), _REMOTE_TOOLS) def test_provider_lists_and_dispatches_a_deployed_mcp_tool(monkeypatch): responses = [ { "choices": [ { "message": { "content": None, "tool_calls": [ { "id": "call-9", "type": "function", "function": {"name": "imageforge__dom_query", "arguments": '{"selector":"#go"}'}, } ], } } ] }, {"choices": [{"message": {"content": "已通过 MCP 工具完成。"}}]}, ] captured: list[dict[str, object]] = [] class Response: def __init__(self, payload): self.payload = payload def __enter__(self): return self def __exit__(self, *_): return False def read(self): return json.dumps(self.payload).encode() def fake_urlopen(request, timeout): captured.append(json.loads(request.data.decode())) return Response(responses.pop(0)) monkeypatch.setattr("mulit_agent_app.infrastructure.openai_compatible_provider.urlopen", fake_urlopen) calls: list[tuple[str, dict]] = [] toolbox = Toolbox("task-1", None, (_endpoint(calls).bridge,)) output = OpenAICompatibleProvider("https://example/v1", "model", _Credentials()).generate("点击按钮", tool_executor=toolbox) assert output == "已通过 MCP 工具完成。" assert calls == [("dom_query", {"selector": "#go"})] # Preset computer tools are always present; the deployed MCP tool is added on top. assert {"run_powershell", "imageforge__dom_query"} <= {tool["function"]["name"] for tool in captured[0]["tools"]} assert captured[1]["messages"][-1] == {"role": "tool", "tool_call_id": "call-9", "content": "MCP 工具结果"} def test_executor_reports_mcp_execution_when_only_a_deployed_tool_was_used(tmp_path): database = AgentDatabase(tmp_path / "agent.sqlite3") calls: list[tuple[str, dict]] = [] def runner(_instruction, _attachments, toolbox): return f"已调用 MCP 工具：{toolbox.call('imageforge__dom_query', {'selector': '#go'})}" executor = TaskExecutor(database, runner) executor.set_mcp_endpoints((_endpoint(calls),)) completed = Event() events: list[dict[str, object]] = [] def capture(event: object) -> None: assert isinstance(event, dict) events.append(event) if event.get("status") == "completed": completed.set() executor.task_event.connect(capture) executor.submit("mcp-task", "在第一张产品图上点击生成按钮") assert completed.wait(3) assert calls == [("dom_query", {"selector": "#go"})] assert events[-1]["execution"] == "mcp" def test_executor_rejects_an_unknown_tool_name_without_failing_the_task(tmp_path): database = AgentDatabase(tmp_path / "agent.sqlite3") def runner(_instruction, _attachments, toolbox): return toolbox.call("imageforge__missing", {}) executor = TaskExecutor(database, runner) executor.set_mcp_endpoints((_endpoint([]),)) completed = Event() events: list[dict[str, object]] = [] executor.task_event.connect( lambda event: (events.append(event), completed.set()) if isinstance(event, dict) and event.get("status") == "completed" else None ) executor.submit("unknown-tool-task", "总结一下") assert completed.wait(3) assert events[-1]["output"] == "工具调用被拒绝：不支持的工具。" def test_http_transport_initialises_keeps_the_session_and_reads_sse(monkeypatch): requests: list[dict] = [] class Headers(dict): pass class Response: def __init__(self, body: str, session_id: str = "") -> None: self._body = body self.headers = Headers({"Mcp-Session-Id": session_id} if session_id else {}) def __enter__(self): return self def __exit__(self, *_): return False def read(self) -> bytes: return self._body.encode() def fake_urlopen(request, timeout): body = json.loads(request.data.decode()) requests.append({"body": body, "session": request.get_header("Mcp-session-id")}) if body["method"] == "initialize": return Response(json.dumps({"jsonrpc": "2.0", "id": 1, "result": {"protocolVersion": "2025-03-26"}}), "session-42") if body["method"] == "tools/list": # Streamable HTTP servers may answer with one SSE data event. payload = {"jsonrpc": "2.0", "id": 2, "result": {"tools": _REMOTE_TOOLS}} return Response(f"event: message\ndata: {json.dumps(payload)}\n\n") return Response(json.dumps({"jsonrpc": "2.0", "id": 3, "result": {"content": [{"type": "text", "text": "已点击"}]}})) monkeypatch.setattr("mulit_agent_app.infrastructure.mcp_client.urlopen", fake_urlopen) client = McpClient(HttpTransport("http://127.0.0.1:9099/mcp"), 30.0) assert client.list_tools() == _REMOTE_TOOLS assert client.call_tool("dom_query", {"selector": "#go"}) == "已点击" assert [item["body"]["method"] for item in requests] == ["initialize", "tools/list", "tools/call"] assert requests[2]["body"]["params"] == {"name": "dom_query", "arguments": {"selector": "#go"}} assert requests[1]["session"] == "session-42" and requests[2]["session"] == "session-42" def test_stdio_transport_spawns_a_server_and_tolerates_stdout_noise(mcp_server_script, python_executable): client = open_client("stdio", command=python_executable, args=(mcp_server_script,)) try: remote_tools = client.list_tools() assert [tool["name"] for tool in remote_tools] == ["echo"] assert client.call_tool("echo", {"text": "你好"}) == "回显：你好" finally: client.close() def test_stdio_transport_reports_a_missing_command_and_stops_a_live_process(python_executable, mcp_server_script): with pytest.raises(McpClientError, match="找不到要执行的命令"): open_client("stdio", command="definitely-not-installed-command") client = open_client("stdio", command=python_executable, args=(mcp_server_script,)) client.list_tools() process = client._transport._process client.close() assert process.poll() is not None, "关闭连接必须结束 MCP 子进程" def test_stdio_tool_is_published_to_the_model_after_deployment(tmp_path, mcp_server_script, python_executable): """The spawn path must produce a tool the model can actually call.""" database = AgentDatabase(tmp_path / "agent.sqlite3") def runner(_instruction, _attachments, toolbox): names = {spec.name for spec in toolbox.specs()} assert "echo__echo" in names return toolbox.call("echo__echo", {"text": "部署成功"}) executor = TaskExecutor(database, runner) client = open_client("stdio", command=python_executable, args=(mcp_server_script,)) executor.set_mcp_endpoints((McpEndpoint("echo", "stdio-key", client, client.list_tools()),)) completed = Event() events: list[dict[str, object]] = [] executor.task_event.connect( lambda event: (events.append(event), completed.set()) if isinstance(event, dict) and event.get("status") == "completed" else None ) try: executor.submit("stdio-task", "回显一句话") assert completed.wait(10) assert events[-1]["output"] == "回显：部署成功" assert events[-1]["execution"] == "mcp" finally: executor.close() assert client._transport._process.poll() is not None def test_replacing_endpoints_closes_the_process_it_dropped(tmp_path, mcp_server_script, python_executable): database = AgentDatabase(tmp_path / "agent.sqlite3") executor = TaskExecutor(database) client = open_client("stdio", command=python_executable, args=(mcp_server_script,)) endpoint = McpEndpoint("echo", "stdio-key", client, client.list_tools()) executor.set_mcp_endpoints((endpoint,)) executor.set_mcp_endpoints(()) assert client._transport._process.poll() is not None assert executor.mcp_endpoints() == ()