You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
255 lines
9.5 KiB
255 lines
9.5 KiB
"""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() == ()
|
|
|