From 9e01cbbec0ef76de90c5be04c50c865aa956e60f Mon Sep 17 00:00:00 2001 From: toot <[email protected]> Date: Wed, 16 Sep 2026 15:31:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E9=A2=84=E8=AE=BE=E6=9C=AC?= =?UTF-8?q?=E6=9C=BA=E5=B7=A5=E5=85=B7=E3=80=81stdio=20MCP=20=E4=BC=A0?= =?UTF-8?q?=E8=BE=93=E4=B8=8E=E5=A4=9A=E8=BD=AE=E4=B8=8A=E4=B8=8B=E6=96=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 预设 12 个本机工具并默认启用，装好即是可操作电脑的 Agent、无需总控配置或安装： 文件与文件夹的查看/读取/写入/建目录/移动/删除（默认不覆盖、删目录需 recursive、 拒绝删除系统目录与用户根目录）、用默认程序打开网址或文件、时间与时区换算 （tzdata 已随包内置）、查看进程、屏幕截图、读写剪贴板。 - 工具下发改为通用：任意 HTTP MCP 服务与 stdio 本机命令（Agent 拉起子进程， 容忍 stdout 噪声，命令缺失或进程崩溃会明确回报），支持多工具并存、 定义未变更时复用连接、替换/删除/退出时回收子进程。 - 多轮上下文：task_dispatch.history 作为此前的 user/assistant 文本拼接到本次请求前， OpenAI 兼容、Anthropic、Gemini 分别适配；Agent 侧独立校验与限幅。 - 工具级实时进度以 progress 事件回传；使用预设工具的任务标记为 builtin 执行来源。 - Skill 与历史内容上限与总控对齐，多个已批准 Skill 不再被静默截断。 - 修复 build.ps1 缺 UTF-8 BOM 导致 build.bat 在 PowerShell 5.1 下语法报错。 --- mulit_agent/build.ps1 | 2 +- mulit_agent/docs/STATUS.md | 9 +- mulit_agent/pyproject.toml | 1 + .../application/agent_service.py | 226 +++++++++- .../application/task_executor.py | 82 +++- .../infrastructure/builtin_tools.py | 423 ++++++++++++++++++ .../infrastructure/controller_session.py | 35 +- .../infrastructure/database.py | 11 + .../infrastructure/mcp_client.py | 270 +++++++++++ .../openai_compatible_provider.py | 53 +-- .../infrastructure/provider_factory.py | 88 ++-- .../infrastructure/tool_registry.py | 241 ++++++++++ mulit_agent/src/mulit_agent_app/main.py | 4 +- mulit_agent/tests/conftest.py | 47 ++ mulit_agent/tests/test_agent_service_mcp.py | 176 +++++++- mulit_agent/tests/test_builtin_tools.py | 149 ++++++ mulit_agent/tests/test_controller_session.py | 30 ++ mulit_agent/tests/test_image_attachments.py | 42 +- mulit_agent/tests/test_mcp_tools.py | 255 +++++++++++ .../tests/test_openai_compatible_provider.py | 36 ++ mulit_agent/tests/test_task_executor.py | 74 +++ 21 files changed, 2110 insertions(+), 144 deletions(-) create mode 100644 mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py create mode 100644 mulit_agent/src/mulit_agent_app/infrastructure/mcp_client.py create mode 100644 mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py create mode 100644 mulit_agent/tests/conftest.py create mode 100644 mulit_agent/tests/test_builtin_tools.py create mode 100644 mulit_agent/tests/test_mcp_tools.py diff --git a/mulit_agent/build.ps1 b/mulit_agent/build.ps1 index c94245a..6e5b456 100644 --- a/mulit_agent/build.ps1 +++ b/mulit_agent/build.ps1 @@ -1,4 +1,4 @@ -$ErrorActionPreference = "Stop" +$ErrorActionPreference = "Stop" $projectRoot = $PSScriptRoot $venvPython = Join-Path $projectRoot ".venv\Scripts\python.exe" diff --git a/mulit_agent/docs/STATUS.md b/mulit_agent/docs/STATUS.md index e10d786..98af789 100644 --- a/mulit_agent/docs/STATUS.md +++ b/mulit_agent/docs/STATUS.md @@ -4,6 +4,11 @@ ## 已完成并验证 +- 已支持总控下发的多轮上下文：`task_dispatch.history` 会作为此前的问答消息（仅 user/assistant 文本，不含工具调用细节）拼接到本次请求之前，OpenAI 兼容、Anthropic、Gemini 三种载荷分别适配（Gemini 使用 user/model 角色）。Agent 侧对历史做独立校验与限幅（最多 100 条、单条 4000 字符、总量 6 万字符），超出时保留最近轮次，异常数据不会影响任务线程。 +- 任务运行可观测：执行期间每次工具调用与返回都会作为 `progress` 事件实时回传给总控（含工具名、参数摘要与结果摘要，单条上限 500 字符，仍走脱敏与离线队列补发），长任务不再“黑盒执行”。任务结束或取消后的迟到进展不会写入；总控侧 `progress` 不改变任务状态，并会顺延“无进展超时”，避免正在跑工具的任务被误判超时。 +- 已预设本机工具，Agent 装好即是“能操作电脑的人”，无需任何总控配置或安装：文件与文件夹的查看/读取/写入/建目录/移动/删除（写与移动默认不覆盖，删除文件夹需显式 recursive，拒绝删除系统目录与用户根目录）、用系统默认程序打开网址或文件、当前时间与时区换算（本机/UTC/UTC+9/IANA，tzdata 已随包内置）、查看进程、屏幕截图（保存 PNG 并返回路径）、读写剪贴板。这些工具由 Agent 本体提供并默认启用，与 `run_powershell` 完全访问能力并存，模型会优先使用这些结构化工具；使用预设工具的任务会回传 `builtin` 执行来源（总控审计为 `task.builtin_tool_execution`），与 完全访问 / MCP 工具 分开可查。 +- 通用工具下发已打通：总控可下发任意 `http` MCP 服务（Agent 以 MCP 客户端连接受控地址，自动 `initialize`、`tools/list` 发现工具，支持 JSON 与 SSE 响应、`Mcp-Session-Id` 会话）、`stdio` 本机命令（Agent 直接拉起子进程，走换行分隔的 JSON-RPC，容忍子进程往 stdout 打印日志，注册表里找不到命令或进程崩溃会明确回报失败原因）以及 `builtin` 内置工具；不再只认 `desktop-filesystem` 一个固定标识。已部署工具会与 `run_powershell` 一起提供给模型调用，调用过 MCP 工具的任务会标记为 `mcp` 执行来源；连接失败的 MCP 服务只使该工具报 `failed`，不会影响任务线程。定义未变更时重连会复用已有连接（不重启子进程），替换或删除工具、退出程序时都会回收拉起的进程。 +- Skill 内容上限与总控对齐（单个 5 万字符、装配前缀 20 万字符），多个已批准 Skill 不再被静默截断。 - 图片附件任务闭环：安全校验总控 `task_dispatch.attachments` 的 MIME、数量与解码大小；附件仅在执行内存传递，OpenAI 兼容、Anthropic、Gemini 分别生成正确图文载荷。无视觉能力的当前模型会安全失败，不会降级为纯文本或影响 WSS 线程。 - 默认完全访问执行器：已认证总控下发的任务可由 Agent 所在 Windows 电脑实际执行文件、程序、网络和桌面应用操作。OpenAI 兼容、Anthropic、Gemini 均支持模型工具调用循环；执行过本机命令的任务会标记为“完全访问”，取消任务会终止仍在跟踪的命令。命令输出、任务结果和审计摘要仍会脱敏，不能保存 API Key、密码、凭据、图片 Base64 或内部推理。 @@ -23,11 +28,11 @@ ## 最近编译 -- `dist/MultiClawAgent/MultiClawAgent.exe` 已于 2026-09-15 重新编译（含单独/批量指派与 5 分钟无进展续期）。 +- `dist/MultiClawAgent/MultiClawAgent.exe` 已于 2026-09-16 重新编译（含 12 个默认启用的本机预设工具、任务运行中的工具级实时进度回传，以及 HTTP/stdio 两种 MCP 传输，tzdata 已内置）。 ## 已规划，尚未实现 -- 总控下发的 Skills、插件和 MCP 工具的版本、权限校验与回滚。 +- 可执行插件包（旧版 Python 插件、脚本型 Skill）的受控安装与隔离运行；当前可通过 `stdio` 工具直接拉起已存在于本机的 MCP 服务或插件宿主，但总控还不能下发插件文件与安装步骤。 - 持久化离线事件外盒、重连补传、任务重启恢复。 - 本机 Agent 守护服务与远程重启、升级。 diff --git a/mulit_agent/pyproject.toml b/mulit_agent/pyproject.toml index 72e6073..0b46fcf 100644 --- a/mulit_agent/pyproject.toml +++ b/mulit_agent/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "websocket-client>=1.8,<2", "pystray>=0.19,<1", "Pillow>=10,<12", + "tzdata>=2024.1", ] [project.optional-dependencies] diff --git a/mulit_agent/src/mulit_agent_app/application/agent_service.py b/mulit_agent/src/mulit_agent_app/application/agent_service.py index 9ed036b..64528e3 100644 --- a/mulit_agent/src/mulit_agent_app/application/agent_service.py +++ b/mulit_agent/src/mulit_agent_app/application/agent_service.py @@ -3,18 +3,106 @@ from __future__ import annotations import hashlib +import json import re import shutil import tempfile +from collections.abc import Iterable from pathlib import Path +from threading import Lock, Thread from urllib.parse import urlparse -from mulit_agent_app.application.task_executor import TaskExecutor +from mulit_agent_app.application.task_executor import MAX_SKILL_CONTENT_CHARS, TaskExecutor from mulit_agent_app.config import AgentConfig from mulit_agent_app.infrastructure.controller_session import ControllerSession from mulit_agent_app.infrastructure.credential_store import CredentialStore from mulit_agent_app.infrastructure.database import AgentDatabase +from mulit_agent_app.infrastructure.mcp_client import McpClient, McpClientError, open_client from mulit_agent_app.infrastructure.provider_factory import build_provider +from mulit_agent_app.infrastructure.tool_registry import McpEndpoint + +_TOOL_ID_PATTERN = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}") +_TOOL_KINDS = {"builtin", "http", "stdio"} +_REMOTE_TOOL_KINDS = {"http", "stdio"} +_DESKTOP_TOOL_ID = "desktop-filesystem" +_MAX_TOOL_ARGUMENTS = 32 +_MAX_TOOL_ARGUMENT_CHARS = 500 + + +def _tool_meta_key(tool_id: str) -> str: + return f"mcp.tool.{tool_id}" + + +def _tool_definition_key(definition: dict) -> str: + """Fingerprint a definition so an unchanged one can keep its live connection.""" + args = definition.get("args") + return "\x1f".join( + [ + str(definition.get("kind", "")), + str(definition.get("endpoint", "")), + str(definition.get("command", "")), + "\x1e".join(str(argument) for argument in args) if isinstance(args, list) else "", + str(definition.get("version", "")), + ] + ) + + +def _validated_tool_definition(tool: dict) -> dict[str, object]: + """Validate one controller-deployed tool before it is stored or applied.""" + tool_id = tool["id"] + version = tool.get("deployment_version") + if not isinstance(version, int) or isinstance(version, bool) or version < 1: + raise ValueError("工具部署版本无效。") + if tool.get("enabled") is not True: + raise ValueError("工具未被启用。") + kind = tool.get("kind") + if kind not in _TOOL_KINDS: + raise ValueError("工具类型无效。") + definition: dict[str, object] = { + "id": tool_id, + "name": str(tool.get("name") or tool_id)[:80], + "kind": kind, + "version": str(tool.get("version") or "1.0.0")[:40], + "deployment_version": version, + } + if kind == "builtin": + if tool_id != _DESKTOP_TOOL_ID: + raise ValueError("Agent 不支持该内置工具。") + if tool.get("scope") != "desktop-directories": + raise ValueError("桌面文件工具配置无效。") + return definition + if kind == "stdio": + command = tool.get("command") + if not isinstance(command, str) or not command.strip(): + raise ValueError("本机命令工具的启动命令不能为空。") + if len(command) > 500 or any(character in command for character in "\x00\r\n"): + raise ValueError("本机命令无效。") + definition["command"] = command.strip() + definition["args"] = _validated_tool_arguments(tool.get("args")) + return definition + endpoint = tool.get("endpoint") + if not isinstance(endpoint, str) or not endpoint.strip(): + raise ValueError("MCP 地址不能为空。") + parsed = urlparse(endpoint.strip()) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise ValueError("MCP 地址必须是有效的 http:// 或 https:// 地址。") + definition["endpoint"] = endpoint.strip().rstrip("/") + return definition + + +def _validated_tool_arguments(value: object) -> list[str]: + if value is None: + return [] + if not isinstance(value, list) or len(value) > _MAX_TOOL_ARGUMENTS: + raise ValueError("本机命令参数无效。") + arguments: list[str] = [] + for argument in value: + if not isinstance(argument, str) or len(argument) > _MAX_TOOL_ARGUMENT_CHARS: + raise ValueError("本机命令参数无效。") + if any(character in argument for character in "\x00\r\n"): + raise ValueError("本机命令参数无效。") + arguments.append(argument) + return arguments class AgentService: @@ -24,9 +112,11 @@ class AgentService: self.session = ControllerSession(config, self._credential_store) self._database = AgentDatabase(config.database_path) self.task_executor = TaskExecutor(self._database) + self._mcp_lock = Lock() + self._mcp_sequence = 0 self._restore_controller_llm_config() self._restore_skill_context() - self.task_executor.set_desktop_filesystem_enabled(self._database.get_meta("mcp.desktop-filesystem.enabled") == "true") + self._restore_mcp_tools() self.session.task_dispatched.connect(self.task_executor.submit) self.session.task_cancelled.connect(self.task_executor.cancel) self.session.config_snapshot_received.connect(self.apply_config_snapshot) @@ -78,6 +168,11 @@ class AgentService: def reconnect(self) -> None: self.session.reconnect() + def close(self) -> None: + """Stop the session and release every deployed MCP connection.""" + self.session.close() + self.task_executor.close() + def apply_config_snapshot(self, payload: object) -> None: if not isinstance(payload, dict): return @@ -141,7 +236,7 @@ class AgentService: version = self._required_text(item, "version") content = self._required_text(item, "content") checksum = self._required_text(item, "checksum") - if len(content) > 50_000 or not re.fullmatch(r"[a-f0-9]{64}", checksum): + if len(content) > MAX_SKILL_CONTENT_CHARS or not re.fullmatch(r"[a-f0-9]{64}", checksum): raise ValueError("Skill 内容或校验值无效。") actual_checksum = hashlib.sha256(content.encode("utf-8")).hexdigest() if actual_checksum != checksum: @@ -175,30 +270,119 @@ class AgentService: self._database.delete_meta_prefix(f"skill.{destination.name}.") def apply_mcp_snapshot(self, payload: object) -> None: + """Apply the controller's complete tool set for this Agent.""" if not isinstance(payload, dict) or not isinstance(payload.get("tools"), list): return + snapshot_ids: set[str] = set() + accepted: list[dict[str, object]] = [] + stored = self._stored_tool_definitions() for tool in payload["tools"]: - if not isinstance(tool, dict) or tool.get("id") != "desktop-filesystem": + if not isinstance(tool, dict): continue + tool_id = tool.get("id") + if not isinstance(tool_id, str) or not _TOOL_ID_PATTERN.fullmatch(tool_id): + continue + snapshot_ids.add(tool_id) version = tool.get("deployment_version") - if not isinstance(version, int) or version < 1: - self.session.send_mcp_ack("desktop-filesystem", 0, "failed", "桌面文件工具部署版本无效。") - return try: - if tool.get("scope") != "desktop-directories" or tool.get("enabled") is not True: - raise ValueError("桌面文件工具配置无效。") - current_version = int(self._database.get_meta("mcp.desktop-filesystem.version") or "0") - if version < current_version: - raise ValueError("拒绝低版本桌面文件工具部署。") - self.task_executor.set_desktop_filesystem_enabled(True) - self._database.set_meta("mcp.desktop-filesystem.enabled", "true") - self._database.set_meta("mcp.desktop-filesystem.version", str(version)) - except (RuntimeError, ValueError) as error: - self.session.send_mcp_ack("desktop-filesystem", version, "failed", str(error)) + definition = _validated_tool_definition(tool) + except ValueError as error: + self.session.send_mcp_ack(tool_id, version if isinstance(version, int) else 0, "failed", str(error)) + continue + previous = stored.get(tool_id) + if previous is not None and int(definition["deployment_version"]) < int(previous["deployment_version"]): + self.session.send_mcp_ack(tool_id, int(definition["deployment_version"]), "failed", "拒绝低版本工具部署。") + continue + accepted.append(definition) + for tool_id in stored: + if tool_id not in snapshot_ids: + self._database.delete_meta(_tool_meta_key(tool_id)) + for definition in accepted: + self._database.set_meta(_tool_meta_key(str(definition["id"])), json.dumps(definition, ensure_ascii=False, separators=(",", ":"))) + self._apply_tool_definitions(self._stored_tool_definitions().values(), acknowledge=True) + + def _stored_tool_definitions(self) -> dict[str, dict[str, object]]: + definitions: dict[str, dict[str, object]] = {} + for key, value in self._database.meta_entries("mcp.tool."): + try: + stored = json.loads(value) + except json.JSONDecodeError: + self._database.delete_meta(key) + continue + if isinstance(stored, dict) and isinstance(stored.get("id"), str): + definitions[str(stored["id"])] = stored + return definitions + + def _restore_mcp_tools(self) -> None: + """Restore the deployed tool set after an Agent restart, without acking.""" + self._apply_tool_definitions(self._stored_tool_definitions().values(), acknowledge=False) + + def _apply_tool_definitions(self, definitions: Iterable[dict[str, object]], acknowledge: bool) -> None: + items = [item for item in definitions if isinstance(item, dict)] + enabled = any(item.get("kind") == "builtin" and item.get("id") == _DESKTOP_TOOL_ID for item in items) + self.task_executor.set_desktop_filesystem_enabled(enabled) + if acknowledge: + for item in items: + if item.get("kind") == "builtin": + self.session.send_mcp_ack(str(item["id"]), int(item["deployment_version"]), "applied") + remote_items = [item for item in items if item.get("kind") in _REMOTE_TOOL_KINDS] + with self._mcp_lock: + self._mcp_sequence += 1 + sequence = self._mcp_sequence + if not remote_items: + self.task_executor.set_mcp_endpoints(()) + return + Thread( + target=self._connect_remote_tools, + args=(remote_items, sequence, acknowledge), + name="agent-mcp-connect", + daemon=True, + ).start() + + def _connect_remote_tools(self, definitions: list[dict[str, object]], sequence: int, acknowledge: bool) -> None: + """Connect to each deployed MCP tool in the background and publish its tools.""" + existing = {endpoint.tool_id: endpoint for endpoint in self.task_executor.mcp_endpoints()} + endpoints: list[McpEndpoint] = [] + for item in definitions: + tool_id, version = str(item["id"]), int(item["deployment_version"]) + definition_key = _tool_definition_key(item) + running = existing.get(tool_id) + if running is not None and running.matches(definition_key): + # An unchanged definition keeps its connection, so a reconnect + # neither respawns a process nor drops an already discovered tool set. + endpoints.append(running) + if acknowledge: + self.session.send_mcp_ack(tool_id, version, "applied") + continue + client: McpClient | None = None + try: + client = open_client( + str(item["kind"]), + str(item.get("endpoint", "")), + str(item.get("command", "")), + tuple(str(argument) for argument in item.get("args") or []), + ) + remote_tools = client.list_tools() + if not remote_tools: + raise McpClientError("MCP 服务没有提供任何工具。") + except (McpClientError, OSError, ValueError) as error: + if client is not None: + client.close() + if acknowledge: + self.session.send_mcp_ack(tool_id, version, "failed", str(error)) + continue + endpoints.append(McpEndpoint(tool_id, definition_key, client, remote_tools)) + if acknowledge: + self.session.send_mcp_ack(tool_id, version, "applied") + with self._mcp_lock: + if sequence != self._mcp_sequence: + # A newer snapshot already superseded this pass: release only the + # connections created here, since reused ones are still in use. + for endpoint in endpoints: + if existing.get(endpoint.tool_id) is not endpoint: + endpoint.close() return - # Repeated snapshots are intentional after reconnect. Reporting the - # same version lets the controller converge without duplicate effects. - self.session.send_mcp_ack("desktop-filesystem", version, "applied") + self.task_executor.set_mcp_endpoints(tuple(endpoints)) def _restore_skill_context(self) -> None: base = self._config.data_dir / "skills" @@ -212,7 +396,7 @@ class AgentService: content = skill_path.read_text(encoding="utf-8") except OSError: continue - parts.append(f"\n--- Skill: {skill_path.parent.name} ---\n{content[:20_000]}") + parts.append(f"\n--- Skill: {skill_path.parent.name} ---\n{content[:MAX_SKILL_CONTENT_CHARS]}") self.task_executor.set_instruction_prefix("\n".join(parts)) @staticmethod diff --git a/mulit_agent/src/mulit_agent_app/application/task_executor.py b/mulit_agent/src/mulit_agent_app/application/task_executor.py index 783bf17..ef19bbe 100644 --- a/mulit_agent/src/mulit_agent_app/application/task_executor.py +++ b/mulit_agent/src/mulit_agent_app/application/task_executor.py @@ -13,6 +13,12 @@ from mulit_agent_app.infrastructure.database import AgentDatabase from mulit_agent_app.infrastructure.desktop_filesystem_tool import DesktopFilesystemTool from mulit_agent_app.infrastructure.full_access_tool import FullAccessTool from mulit_agent_app.infrastructure.openai_compatible_provider import visible_model_output +from mulit_agent_app.infrastructure.tool_registry import McpEndpoint, Toolbox + +# One Skill document is bounded by the same limit the controller enforces, and the +# assembled prefix leaves room for several approved Skills without silent truncation. +MAX_SKILL_CONTENT_CHARS = 50_000 +MAX_SKILL_PREFIX_CHARS = 200_000 class TaskExecutor: @@ -31,6 +37,7 @@ class TaskExecutor: self._desktop_tool = DesktopFilesystemTool() self._full_access_tool = FullAccessTool() self._full_access_enabled = True + self._mcp_endpoints: tuple[McpEndpoint, ...] = () self.task_event = EventHook() def set_runner(self, runner: Callable[[str, tuple[TaskAttachment, ...]], str], supports_images: bool = False) -> None: @@ -42,11 +49,28 @@ class TaskExecutor: def set_instruction_prefix(self, value: str) -> None: """Install controller-approved Skill guidance for later model requests.""" with self._lock: - self._instruction_prefix = value[:40_000] + self._instruction_prefix = value[:MAX_SKILL_PREFIX_CHARS] def set_desktop_filesystem_enabled(self, enabled: bool) -> None: self._desktop_tool.set_enabled(enabled) + def set_mcp_endpoints(self, endpoints: tuple[McpEndpoint, ...]) -> None: + """Install the controller-deployed MCP tools and stop the ones that were replaced.""" + with self._lock: + previous, self._mcp_endpoints = self._mcp_endpoints, tuple(endpoints) + live = {id(endpoint) for endpoint in self._mcp_endpoints} + for endpoint in previous: + if id(endpoint) not in live: + endpoint.close() + + def mcp_endpoints(self) -> tuple[McpEndpoint, ...]: + with self._lock: + return self._mcp_endpoints + + def close(self) -> None: + """Release every deployed MCP connection, including spawned processes.""" + self.set_mcp_endpoints(()) + def set_full_access_enabled(self, enabled: bool) -> None: """Keep the historical narrow tool available only for legacy Agents.""" self._full_access_enabled = enabled @@ -56,7 +80,14 @@ class TaskExecutor: for task_id in self._database.recover_interrupted_tasks(self._now()): self.task_event.emit({"task_id": task_id, "status": "failed", "message": "Agent 重启后已安全结束未完成任务。", "error": "Agent 重启前的任务无法安全恢复。"}) - def submit(self, task_id: str, instruction: str, deadline_at: str = "", attachments: tuple[TaskAttachment, ...] = ()) -> None: + def submit( + self, + task_id: str, + instruction: str, + deadline_at: str = "", + attachments: tuple[TaskAttachment, ...] = (), + history: tuple[tuple[str, str], ...] = (), + ) -> None: existing_status = self._database.task_status(task_id) if existing_status is not None: self.task_event.emit( @@ -65,7 +96,7 @@ class TaskExecutor: return self._database.create_task(task_id, instruction, self._now()) self.task_event.emit({"task_id": task_id, "status": "accepted", "message": "任务已接收。"}) - self._schedule(task_id, 0.08, lambda: self._start(task_id, instruction, attachments)) + self._schedule(task_id, 0.08, lambda: self._start(task_id, instruction, attachments, history)) if deadline_at: try: deadline = datetime.fromisoformat(deadline_at) @@ -76,7 +107,7 @@ class TaskExecutor: self._deadline_windows[task_id] = seconds self._renew_deadline(task_id, seconds) - def _start(self, task_id: str, instruction: str, attachments: tuple[TaskAttachment, ...]) -> None: + def _start(self, task_id: str, instruction: str, attachments: tuple[TaskAttachment, ...], history: tuple[tuple[str, str], ...] = ()) -> None: if self._database.task_status(task_id) != "accepted": return with self._lock: @@ -102,7 +133,7 @@ class TaskExecutor: self._schedule(task_id, 1.5, lambda: self._complete_demo(task_id, instruction)) return prompt = f"{prefix}\n\n用户任务：\n{instruction}" if prefix else instruction - thread = Thread(target=self._run_model, args=(task_id, prompt, attachments, runner), daemon=True) + thread = Thread(target=self._run_model, args=(task_id, prompt, attachments, history, runner), daemon=True) thread.start() def _complete_demo(self, task_id: str, instruction: str) -> None: @@ -111,13 +142,15 @@ class TaskExecutor: output = f"内置演示执行器已完成任务：{instruction}" self._finish(task_id, "completed", "任务已完成。", output=output, execution_source="demo") - def _run_model(self, task_id: str, instruction: str, attachments: tuple[TaskAttachment, ...], runner: Callable[[str, tuple[TaskAttachment, ...]], str]) -> None: - used_full_access = False - - def run_local(command: str) -> str: - nonlocal used_full_access - used_full_access = True - return self._full_access_tool.execute(task_id, command) + def _run_model(self, task_id: str, instruction: str, attachments: tuple[TaskAttachment, ...], history: tuple[tuple[str, str], ...], runner: Callable[..., str]) -> None: + with self._lock: + endpoints = self._mcp_endpoints + toolbox = Toolbox( + task_id, + self._full_access_tool, + tuple(endpoint.bridge for endpoint in endpoints), + on_progress=self._report_progress(task_id), + ) try: staged_paths = self._full_access_tool.stage_attachments(task_id, attachments) @@ -131,8 +164,10 @@ class TaskExecutor: output = runner(instruction) elif parameter_count == 2: output = runner(instruction, attachments) + elif parameter_count == 3: + output = runner(instruction, attachments, toolbox) else: - output = runner(instruction, attachments, run_local) + output = runner(instruction, attachments, toolbox, history) except Exception as error: # noqa: BLE001 - provider plug-ins must not take down the Agent process. if self._database.task_status(task_id) == "running": detail = str(error).strip() or "未知错误。" @@ -141,7 +176,7 @@ class TaskExecutor: finally: self._full_access_tool.cleanup_task_attachments(task_id) if self._database.task_status(task_id) == "running": - if not used_full_access and self._requires_local_action(instruction): + if not toolbox.used_any_tool and self._requires_local_action(instruction): self._finish( task_id, "failed", @@ -149,9 +184,26 @@ class TaskExecutor: error="任务要求本机操作，但模型只返回了文字，未执行任何操作。", ) return - execution = "full-access" if used_full_access else "llm" + if toolbox.used_full_access: + execution = "full-access" + elif toolbox.used_builtin: + execution = "builtin" + elif toolbox.used_mcp: + execution = "mcp" + else: + execution = "llm" self._finish(task_id, "completed", "任务已完成。", output=visible_model_output(output), execution_source=execution) + def _report_progress(self, task_id: str) -> Callable[[str], None]: + """Stream one intermediate step so a long task is observable while it runs.""" + + def report(message: str) -> None: + if self._database.task_status(task_id) not in {"accepted", "running"}: + return # A finished or cancelled task must not gain new steps. + self.task_event.emit({"task_id": task_id, "status": "progress", "message": message}) + + return report + def cancel(self, task_id: str, reason: str = "任务已由总控取消。") -> None: if self._database.task_status(task_id) not in {"accepted", "running"}: return diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py b/mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py new file mode 100644 index 0000000..efc06c0 --- /dev/null +++ b/mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py @@ -0,0 +1,423 @@ +"""Preset computer tools that every Agent has without any controller configuration. + +A freshly installed Agent must already behave like someone who can operate the +computer it runs on. These tools therefore ship inside the Agent, need no +installation on the Agent machine and are always enabled. They are written with +the standard library only, except for the three capabilities Windows itself +exposes through PowerShell (process list, screenshot, clipboard). +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import sys +import webbrowser +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta, timezone +from pathlib import Path +from uuid import uuid4 +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +_MAX_RESULT_CHARS = 12_000 +_MAX_TEXT_CHARS = 200_000 +_MAX_LIST_ENTRIES = 500 +_POWERSHELL_TIMEOUT = 60.0 +_WEEKDAYS = ("星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日") + +# Deleting these would break the machine rather than clean up the user's work. +_PROTECTED_DELETE_TARGETS = ( + "C:\\Windows", + "C:\\Program Files", + "C:\\Program Files (x86)", + "C:\\ProgramData", +) + + +@dataclass(frozen=True) +class BuiltinToolSpec: + """One preset tool exposed to the model without any controller deployment.""" + + name: str + description: str + parameters: dict + handler: Callable[[dict], str] + + +def _text(arguments: dict, key: str) -> str: + value = arguments.get(key) + if not isinstance(value, str): + raise TypeError(f"参数 {key} 必须是字符串。") + if not value.strip(): + raise ValueError(f"参数 {key} 不能为空。") + return value.strip() + + +def _flag(arguments: dict, key: str) -> bool: + value = arguments.get(key, False) + if not isinstance(value, bool): + raise TypeError(f"参数 {key} 必须是布尔值。") + return value + + +def _count(arguments: dict, key: str, default: int, maximum: int) -> int: + value = arguments.get(key, default) + if isinstance(value, bool) or not isinstance(value, int): + raise TypeError(f"参数 {key} 必须是整数。") + return max(1, min(maximum, value)) + + +def _path(arguments: dict, key: str = "path") -> Path: + return Path(_text(arguments, key)).expanduser() + + +def _run_powershell(script: str, environment: dict[str, str] | None = None) -> str: + """Run a fixed script. Model-supplied text is passed through the environment + so it can never be interpreted as PowerShell code.""" + process = subprocess.run( + [ + "powershell.exe", + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-ExecutionPolicy", + "Bypass", + "-Command", + "$OutputEncoding = [Console]::OutputEncoding = [Text.Encoding]::UTF8; " + script, + ], + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=_POWERSHELL_TIMEOUT, + check=False, + env={**os.environ, **(environment or {})}, + ) + output = "\n".join(part for part in ((process.stdout or "").strip(), (process.stderr or "").strip()) if part) + if process.returncode: + raise RuntimeError(f"系统命令返回 {process.returncode}：{output[:500] or '无输出'}") + return output[:_MAX_RESULT_CHARS] or "已完成，没有文本输出。" + + +def _list_directory(arguments: dict) -> str: + target = _path(arguments) + if not target.is_dir(): + raise ValueError(f"不是可读取的文件夹：{target}") + lines = [f"{target}"] + entries = sorted(target.iterdir(), key=lambda item: (item.is_file(), item.name.casefold())) + for index, item in enumerate(entries): + if index >= _MAX_LIST_ENTRIES: + lines.append(f"… 其余 {len(entries) - _MAX_LIST_ENTRIES} 项已省略") + break + if item.is_dir(): + lines.append(f"[文件夹] {item.name}") + continue + try: + size = item.stat().st_size + except OSError: + size = 0 + lines.append(f"[文件] {item.name} {size} 字节") + if len(lines) == 1: + lines.append("（空文件夹）") + return "\n".join(lines) + + +def _read_text_file(arguments: dict) -> str: + target = _path(arguments) + if not target.is_file(): + raise ValueError(f"不是可读取的文件：{target}") + size = target.stat().st_size + if size > _MAX_TEXT_CHARS: + raise ValueError(f"文件过大（{size} 字节），请改用 run_powershell 分段读取。") + content = target.read_text(encoding="utf-8", errors="replace") + return f"{target}\n\n{content}" + + +def _write_text_file(arguments: dict) -> str: + target = _path(arguments) + content = arguments.get("content") + if not isinstance(content, str): + raise TypeError("参数 content 必须是字符串。") + if target.exists() and not _flag(arguments, "overwrite"): + raise ValueError("目标已存在；确实要覆盖时请传 overwrite=true。") + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + return f"已写入 {len(content)} 个字符：{target}" + + +def _create_directory(arguments: dict) -> str: + target = _path(arguments) + existed = target.exists() + target.mkdir(parents=True, exist_ok=True) + return f"{'文件夹已存在' if existed else '已创建文件夹'}：{target}" + + +def _move_path(arguments: dict) -> str: + source = _path(arguments, "source") + destination = _path(arguments, "destination") + if not source.exists(): + raise ValueError(f"源路径不存在：{source}") + if destination.exists(): + if not _flag(arguments, "overwrite"): + raise ValueError("目标已存在；确实要替换时请传 overwrite=true。") + if destination.is_dir(): + raise ValueError("拒绝用覆盖方式替换已存在的文件夹，请先明确删除。") + destination.unlink() + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(source), str(destination)) + return f"已移动：{source} → {destination}" + + +def _is_protected(path: Path) -> bool: + resolved = path.resolve() + if len(str(resolved)) <= 3: # A drive root such as C:\ + return True + if resolved == Path.home().resolve(): + return True + return any(resolved == Path(protected).resolve() or resolved.is_relative_to(Path(protected)) for protected in _PROTECTED_DELETE_TARGETS) + + +def _delete_path(arguments: dict) -> str: + target = _path(arguments) + if not target.exists(): + raise ValueError(f"路径不存在：{target}") + if _is_protected(target): + raise ValueError(f"拒绝删除系统或用户根目录：{target}") + if target.is_dir(): + if not _flag(arguments, "recursive"): + raise ValueError("删除文件夹需要同时传 recursive=true。") + shutil.rmtree(target) + else: + target.unlink() + return f"已删除：{target}" + + +def _open_target(arguments: dict) -> str: + target = _text(arguments, "target") + lowered = target.casefold() + if lowered.startswith(("http://", "https://")): + if not webbrowser.open(target): + raise RuntimeError(f"系统未能打开网址：{target}") + return f"已用系统默认程序打开网址：{target}" + local = Path(target).expanduser() + if not local.exists(): + raise ValueError("只支持 http/https 网址，或本机上已存在的文件/文件夹。") + opener = getattr(os, "startfile", None) + if opener is None: + raise RuntimeError("当前系统不支持打开本地路径。") + opener(str(local)) + return f"已用系统默认程序打开：{local}" + + +def _resolve_timezone(value: str): + normalized = value.strip() + if normalized.casefold() in {"", "local", "本机", "本地"}: + return datetime.now().astimezone().tzinfo + if normalized.casefold() in {"utc", "gmt"}: + return UTC + offset = _parse_utc_offset(normalized) + if offset is not None: + return offset + try: + return ZoneInfo(normalized) + except (KeyError, ZoneInfoNotFoundError, ValueError) as error: + raise ValueError(f"无法识别时区 {value}；请使用 IANA 名称（如 Asia/Tokyo）或 UTC+9 形式。") from error + + +def _parse_utc_offset(normalized: str) -> timezone | None: + if not normalized.upper().startswith(("UTC", "GMT")): + return None + digits = normalized[3:].replace(":", "") + if len(digits) < 2: + return None + sign, digits = digits[0], digits[1:] + if sign not in {"+", "-"} or not digits.isdigit(): + return None + hours = int(digits[:2]) + minutes = int(digits[2:4]) if len(digits) >= 4 else 0 + if hours > 14 or minutes > 59: + return None + delta = timedelta(hours=hours, minutes=minutes) + return timezone(-delta if sign == "-" else delta) + + +def _current_time(arguments: dict) -> str: + requested = arguments.get("timezone", "local") + if not isinstance(requested, str): + raise TypeError("参数 timezone 必须是字符串。") + moment = datetime.now(_resolve_timezone(requested)) + weekday = _WEEKDAYS[moment.weekday()] + return f"{moment:%Y-%m-%d %H:%M:%S}（{weekday}，{moment.tzname() or requested}）" + + +def _list_processes(arguments: dict) -> str: + limit = _count(arguments, "limit", 20, 100) + name_filter = arguments.get("name", "") + if not isinstance(name_filter, str): + raise TypeError("参数 name 必须是字符串。") + script = ( + f"$items = Get-Process | Sort-Object -Property WS -Descending;" + f"if ($env:MULTICLAW_PROCESS_FILTER) {{ $items = $items | Where-Object {{ $_.Name -like ('*' + $env:MULTICLAW_PROCESS_FILTER + '*') }} }};" + f"$items | Select-Object -First {limit} Name, Id, @{{n='内存MB';e={{[int]($_.WS/1MB)}}}} | Format-Table -AutoSize | Out-String -Width 120" + ) + return _run_powershell(script, {"MULTICLAW_PROCESS_FILTER": name_filter.strip()}) + + +def _screenshot(arguments: dict) -> str: + requested = arguments.get("path") + if requested is None: + directory = Path(os.environ.get("TEMP", str(Path.home()))) / "MultiClawAgent" / "screenshots" + directory.mkdir(parents=True, exist_ok=True) + target = directory / f"screen-{datetime.now().astimezone():%Y%m%d-%H%M%S}-{uuid4().hex[:6]}.png" + elif isinstance(requested, str) and requested.strip(): + target = Path(requested.strip()).expanduser() + target.parent.mkdir(parents=True, exist_ok=True) + else: + raise ValueError("参数 path 必须是文件路径。") + script = ( + "Add-Type -AssemblyName System.Windows.Forms, System.Drawing;" + "$area = [System.Windows.Forms.SystemInformation]::VirtualScreen;" + "$bitmap = New-Object System.Drawing.Bitmap $area.Width, $area.Height;" + "$graphics = [System.Drawing.Graphics]::FromImage($bitmap);" + "$graphics.CopyFromScreen($area.Left, $area.Top, 0, 0, $bitmap.Size);" + "$bitmap.Save($env:MULTICLAW_SCREENSHOT_PATH, [System.Drawing.Imaging.ImageFormat]::Png);" + "$graphics.Dispose(); $bitmap.Dispose();" + "Write-Output $env:MULTICLAW_SCREENSHOT_PATH" + ) + _run_powershell(script, {"MULTICLAW_SCREENSHOT_PATH": str(target)}) + if not target.is_file(): + raise RuntimeError("截图没有生成，请检查该电脑是否有可用桌面会话。") + return f"已保存屏幕截图：{target}" + + +def _read_clipboard(_: dict) -> str: + content = _run_powershell("Get-Clipboard -Raw") + return f"剪贴板文本：\n{content}" + + +def _write_clipboard(arguments: dict) -> str: + content = arguments.get("text") + if not isinstance(content, str): + raise TypeError("参数 text 必须是字符串。") + _run_powershell("Set-Clipboard -Value $env:MULTICLAW_CLIPBOARD_TEXT", {"MULTICLAW_CLIPBOARD_TEXT": content}) + return f"已写入剪贴板：{len(content)} 个字符" + + +def _object_schema(properties: dict, required: list[str] | None = None) -> dict: + return {"type": "object", "properties": properties, "required": required or [], "additionalProperties": False} + + +def builtin_tools() -> tuple[BuiltinToolSpec, ...]: + """The preset tool set: a ready-to-use operator of its own computer.""" + return ( + BuiltinToolSpec( + "computer_list_directory", + "列出本机某个文件夹中的文件和子文件夹（含大小）。路径可以是绝对路径或 ~ 开头。", + _object_schema({"path": {"type": "string", "description": "文件夹路径。"}}, ["path"]), + _list_directory, + ), + BuiltinToolSpec( + "computer_read_text_file", + "读取本机一个 UTF-8 文本文件的内容。超过 20 万字符的文件会被拒绝。", + _object_schema({"path": {"type": "string", "description": "文件路径。"}}, ["path"]), + _read_text_file, + ), + BuiltinToolSpec( + "computer_write_text_file", + "在本机写入一个 UTF-8 文本文件（父目录不存在会自动创建）。目标已存在时必须显式传 overwrite=true。", + _object_schema( + { + "path": {"type": "string", "description": "文件路径。"}, + "content": {"type": "string", "description": "要写入的完整文本。"}, + "overwrite": {"type": "boolean", "description": "目标已存在时是否覆盖。", "default": False}, + }, + ["path", "content"], + ), + _write_text_file, + ), + BuiltinToolSpec( + "computer_create_directory", + "在本机创建一个文件夹（含缺失的父目录）。", + _object_schema({"path": {"type": "string", "description": "文件夹路径。"}}, ["path"]), + _create_directory, + ), + BuiltinToolSpec( + "computer_move_path", + "在本机移动或重命名文件/文件夹。目标已存在时必须显式传 overwrite=true 且只允许覆盖文件。", + _object_schema( + { + "source": {"type": "string", "description": "源路径。"}, + "destination": {"type": "string", "description": "目标路径。"}, + "overwrite": {"type": "boolean", "description": "目标已存在时是否替换。", "default": False}, + }, + ["source", "destination"], + ), + _move_path, + ), + BuiltinToolSpec( + "computer_delete_path", + "删除本机的一个文件或文件夹。删除文件夹必须同时传 recursive=true；系统目录和用户根目录会被拒绝。", + _object_schema( + { + "path": {"type": "string", "description": "要删除的路径。"}, + "recursive": {"type": "boolean", "description": "删除文件夹时是否连同内容一起删除。", "default": False}, + }, + ["path"], + ), + _delete_path, + ), + BuiltinToolSpec( + "computer_open", + "用本机系统默认程序打开一个网址（http/https）或已存在的文件/文件夹，例如打开浏览器、打开文档。", + _object_schema({"target": {"type": "string", "description": "网址或本机路径。"}}, ["target"]), + _open_target, + ), + BuiltinToolSpec( + "computer_time", + "查询当前时间。timezone 可填 local（本机时区）、UTC、UTC+9 形式，或 IANA 名称如 Asia/Tokyo、America/New_York。", + _object_schema({"timezone": {"type": "string", "description": "时区，默认 local。", "default": "local"}}), + _current_time, + ), + BuiltinToolSpec( + "computer_list_processes", + "列出本机正在运行的进程（按内存占用排序），可用 name 做模糊过滤。", + _object_schema( + { + "limit": {"type": "integer", "description": "返回条数，默认 20，最多 100。"}, + "name": {"type": "string", "description": "按进程名模糊过滤，可留空。"}, + } + ), + _list_processes, + ), + BuiltinToolSpec( + "computer_screenshot", + "截取本机当前屏幕并保存为 PNG；不传 path 时保存到本机临时目录。返回保存路径。", + _object_schema({"path": {"type": "string", "description": "可选的保存路径。"}}), + _screenshot, + ), + BuiltinToolSpec( + "computer_read_clipboard", + "读取本机剪贴板中的文本。", + _object_schema({}), + _read_clipboard, + ), + BuiltinToolSpec( + "computer_write_clipboard", + "把一段文本写入本机剪贴板。", + _object_schema({"text": {"type": "string", "description": "要写入的文本。"}}, ["text"]), + _write_clipboard, + ), + ) + + +def builtin_tool_names() -> tuple[str, ...]: + return tuple(spec.name for spec in builtin_tools()) + + +if __name__ == "__main__": # A quick manual check on the Agent machine. + print("平台:", sys.platform) + for spec in builtin_tools(): + print("-", spec.name) diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py b/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py index 327a380..1e8edf0 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py @@ -19,6 +19,12 @@ from mulit_agent_app.events import EventHook from mulit_agent_app.infrastructure.credential_store import CredentialStore from mulit_agent_app.infrastructure.database import AgentDatabase +# A follow-up turn carries plain user/assistant text only. These bounds mirror the +# controller so a malformed payload can never grow one request without limit. +_MAX_HISTORY_TURNS = 100 +_MAX_HISTORY_CHARS = 60_000 +_MAX_HISTORY_TURN_CHARS = 4_000 + class ControllerSession: _RECONNECT_DELAYS = (2, 5, 10, 30) @@ -264,7 +270,34 @@ class ControllerSession: self.connection_error.emit(f"总控下发的任务图片无效，已拒绝：{error}") return deadline_at = payload.get("deadline_at") - self.task_dispatched.emit(task_id, instruction, deadline_at if isinstance(deadline_at, str) else "", attachments) + history = self._validated_history(payload.get("history", [])) + self.task_dispatched.emit(task_id, instruction, deadline_at if isinstance(deadline_at, str) else "", attachments, history) + + @staticmethod + def _validated_history(value: object) -> tuple[tuple[str, str], ...]: + """Keep the newest earlier turns that fit the budget, oldest first. + + Only plain user/assistant text is accepted: tool transcripts never travel + between the controller and an Agent. + """ + if not isinstance(value, list): + return () + accepted: list[tuple[str, str]] = [] + budget = _MAX_HISTORY_CHARS + for item in reversed(value): + if len(accepted) >= _MAX_HISTORY_TURNS: + break + if not isinstance(item, dict): + continue + role, content = item.get("role"), item.get("content") + if role not in {"user", "assistant"} or not isinstance(content, str) or not content.strip(): + continue + content = content.strip()[:_MAX_HISTORY_TURN_CHARS] + budget -= len(content) + if budget < 0: + break + accepted.append((role, content)) + return tuple(reversed(accepted)) @staticmethod def _validated_attachments(value: object) -> tuple[TaskAttachment, ...]: diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/database.py b/mulit_agent/src/mulit_agent_app/infrastructure/database.py index 1fb91d6..f011834 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/database.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/database.py @@ -96,6 +96,17 @@ class AgentDatabase: with sqlite3.connect(self._database_path) as connection: connection.execute("DELETE FROM application_meta WHERE key LIKE ?", (prefix + "%",)) + def delete_meta(self, key: str) -> None: + with sqlite3.connect(self._database_path) as connection: + connection.execute("DELETE FROM application_meta WHERE key = ?", (key,)) + + def meta_entries(self, prefix: str) -> list[tuple[str, str]]: + with sqlite3.connect(self._database_path) as connection: + rows = connection.execute( + "SELECT key, value FROM application_meta WHERE key LIKE ? ORDER BY key", (prefix + "%",) + ).fetchall() + return [(str(row[0]), str(row[1])) for row in rows] + def enqueue_event(self, payload: str, created_at: str) -> None: with sqlite3.connect(self._database_path) as connection: connection.execute("INSERT INTO event_outbox(payload, created_at) VALUES (?, ?)", (payload, created_at)) diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/mcp_client.py b/mulit_agent/src/mulit_agent_app/infrastructure/mcp_client.py new file mode 100644 index 0000000..02c50a4 --- /dev/null +++ b/mulit_agent/src/mulit_agent_app/infrastructure/mcp_client.py @@ -0,0 +1,270 @@ +"""MCP client and transports for controller-deployed tools. + +The controller never executes a tool: it only deploys a definition. This module +turns that definition into real model-callable tools on the Agent host, over +either Streamable HTTP or a locally spawned stdio server process. +""" + +from __future__ import annotations + +import json +import shutil +import subprocess +import sys +import time +from collections import deque +from json import JSONDecodeError +from queue import Empty, Queue +from threading import Lock, Thread +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + +PROTOCOL_VERSION = "2025-03-26" + +_INITIALISE_TIMEOUT = 30.0 +_HTTP_CALL_TIMEOUT = 30.0 +_STDIO_CALL_TIMEOUT = 120.0 +_MAX_RESULT_CHARS = 12_000 +_SHUTDOWN = object() + + +class McpClientError(RuntimeError): + """A safe message that may be reported back to the controller.""" + + +def parse_message(body: str) -> dict: + """Read a JSON body or the single SSE data event a Streamable-HTTP server may send.""" + try: + payload = json.loads(body) + except JSONDecodeError: + payload = None + if isinstance(payload, dict): + return payload + for line in body.splitlines(): + if not line.startswith("data:"): + continue + try: + event = json.loads(line[5:].strip()) + except JSONDecodeError: + continue + if isinstance(event, dict): + return event + raise McpClientError("MCP 服务返回了无法识别的响应。") + + +class HttpTransport: + """Streamable-HTTP transport, including MCP session handling.""" + + def __init__(self, endpoint: str) -> None: + self._endpoint = endpoint + self._session_id = "" + + def request(self, payload: dict, timeout: float) -> dict: + headers = {"Accept": "application/json, text/event-stream", "Content-Type": "application/json"} + if self._session_id: + headers["Mcp-Session-Id"] = self._session_id + request = Request( + self._endpoint, + data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), + headers=headers, + method="POST", + ) + try: + with urlopen(request, timeout=timeout) as response: + session_id = response.headers.get("Mcp-Session-Id") + if session_id: + self._session_id = session_id + body = response.read().decode("utf-8", errors="replace") + except HTTPError as error: + raise McpClientError(f"MCP 服务返回 HTTP {error.code}。") from error + except URLError as error: + raise McpClientError(f"无法连接 MCP 服务：{error.reason}") from error + except OSError as error: + raise McpClientError(f"无法连接 MCP 服务：{error}") from error + return parse_message(body) + + def close(self) -> None: + return None + + +class StdioTransport: + """Newline-delimited JSON-RPC over a locally spawned MCP server process.""" + + def __init__(self, command: str, args: tuple[str, ...] = ()) -> None: + executable = shutil.which(command) + if executable is None: + raise McpClientError(f"找不到要执行的命令：{command}") + self._messages: Queue = Queue() + self._diagnostics: deque[str] = deque(maxlen=20) + creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) if sys.platform == "win32" else 0 + try: + self._process = subprocess.Popen( + [executable, *args], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + bufsize=1, + creationflags=creation_flags, + ) + except OSError as error: + raise McpClientError(f"无法启动 MCP 进程：{error}") from error + Thread(target=self._read_stdout, daemon=True, name="agent-mcp-stdout").start() + Thread(target=self._drain_stderr, daemon=True, name="agent-mcp-stderr").start() + + def request(self, payload: dict, timeout: float) -> dict: + stream = self._process.stdin + if stream is None or self._process.poll() is not None: + raise McpClientError(f"MCP 进程已退出。{self._recent_output()}") + try: + stream.write(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n") + stream.flush() + except (OSError, ValueError) as error: + raise McpClientError(f"无法向 MCP 进程写入请求：{error}") from error + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise McpClientError(f"MCP 进程未在 {timeout:.0f} 秒内响应。{self._recent_output()}") + try: + message = self._messages.get(timeout=remaining) + except Empty: + raise McpClientError(f"MCP 进程未在 {timeout:.0f} 秒内响应。{self._recent_output()}") from None + if message is _SHUTDOWN: + raise McpClientError(f"MCP 进程已结束。{self._recent_output()}") + if message.get("id") == payload.get("id"): + return message + + def close(self) -> None: + process = self._process + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + pass + for stream in (process.stdin, process.stdout, process.stderr): + if stream is not None: + try: + stream.close() + except OSError: + pass + + def _read_stdout(self) -> None: + stream = self._process.stdout + if stream is None: + self._messages.put(_SHUTDOWN) + return + for line in stream: + line = line.strip() + if not line: + continue + try: + message = json.loads(line) + except JSONDecodeError: + # A server that logs to stdout must not break the protocol stream. + self._diagnostics.append(f"忽略非协议输出：{line[:160]}") + continue + if isinstance(message, dict) and message.get("id") is not None: + self._messages.put(message) + self._messages.put(_SHUTDOWN) + + def _drain_stderr(self) -> None: + stream = self._process.stderr + if stream is None: + return + for line in stream: + line = line.strip() + if line: + self._diagnostics.append(line[:160]) + + def _recent_output(self) -> str: + if not self._diagnostics: + return "" + return "最近输出：" + " | ".join(list(self._diagnostics)[-3:]) + + +class McpClient: + """The MCP protocol layer; a transport only moves JSON-RPC messages.""" + + def __init__(self, transport: HttpTransport | StdioTransport, call_timeout: float) -> None: + self._transport = transport + self._call_timeout = call_timeout + self._initialised = False + self._next_id = 1 + self._lock = Lock() + + def list_tools(self) -> list[dict]: + result = self._request("tools/list", None) + tools = result.get("tools") + if not isinstance(tools, list): + raise McpClientError("MCP 服务返回了无法识别的工具清单。") + return [tool for tool in tools if isinstance(tool, dict) and isinstance(tool.get("name"), str) and tool["name"]] + + def call_tool(self, name: str, arguments: dict) -> str: + result = self._request("tools/call", {"name": name, "arguments": arguments}, self._call_timeout) + content = result.get("content") + if isinstance(content, list): + text = "\n".join( + str(item.get("text", "")) for item in content if isinstance(item, dict) and item.get("type") == "text" + ).strip() + if text: + return text[: _MAX_RESULT_CHARS] + return json.dumps(result, ensure_ascii=False)[: _MAX_RESULT_CHARS] + + def close(self) -> None: + self._transport.close() + + def _request(self, method: str, params: dict | None, timeout: float = _INITIALISE_TIMEOUT) -> dict: + with self._lock: + self._initialise() + message = self._transport.request( + {"jsonrpc": "2.0", "id": self._take_id(), "method": method, "params": params or {}}, + timeout, + ) + if "error" in message: + error = message["error"] + detail = error.get("message", str(error)) if isinstance(error, dict) else str(error) + raise McpClientError(f"MCP 服务返回错误：{detail}") + result = message.get("result") + if not isinstance(result, dict): + raise McpClientError("MCP 服务没有返回结果。") + return result + + def _initialise(self) -> None: + if self._initialised: + return + self._transport.request( + { + "jsonrpc": "2.0", + "id": self._take_id(), + "method": "initialize", + "params": { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": {"name": "multiclaw-agent", "version": "1.0.0"}, + }, + }, + _INITIALISE_TIMEOUT, + ) + self._initialised = True + + def _take_id(self) -> int: + request_id = self._next_id + self._next_id += 1 + return request_id + + +def open_client(kind: str, endpoint: str = "", command: str = "", args: tuple[str, ...] = ()) -> McpClient: + """Build the client for one deployed tool definition.""" + if kind == "http": + return McpClient(HttpTransport(endpoint), _HTTP_CALL_TIMEOUT) + if kind == "stdio": + return McpClient(StdioTransport(command, args), _STDIO_CALL_TIMEOUT) + raise McpClientError(f"不支持的 MCP 工具类型：{kind}") diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py b/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py index bcae332..b7b3a20 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py @@ -4,12 +4,12 @@ from __future__ import annotations import json import re -from collections.abc import Callable from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen from mulit_agent_app.domain.attachments import TaskAttachment from mulit_agent_app.infrastructure.credential_store import CredentialStore +from mulit_agent_app.infrastructure.tool_registry import as_toolbox, system_prompt class ProviderError(RuntimeError): @@ -38,7 +38,8 @@ class OpenAICompatibleProvider: self, instruction: str, attachments: tuple[TaskAttachment, ...] = (), - tool_executor: Callable[[str], str] | None = None, + tool_executor: object = None, + history: tuple[tuple[str, str], ...] = (), ) -> str: if not self._model: raise ProviderError("总控尚未为此 Agent 下发 LLM 模型配置。") @@ -49,34 +50,20 @@ class OpenAICompatibleProvider: if attachments: content = [{"type": "text", "text": instruction}] content.extend({"type": "image_url", "image_url": {"url": attachment.data_url}} for attachment in attachments) - messages: list[dict[str, object]] = [{"role": "user", "content": content}] + toolbox = as_toolbox(tool_executor) + messages: list[dict[str, object]] = [] + if toolbox is not None: + messages.append({"role": "system", "content": system_prompt(toolbox)}) + messages.extend({"role": role, "content": text} for role, text in history) + messages.append({"role": "user", "content": content}) request_body: dict[str, object] = {"model": self._model, "messages": messages, "stream": False} - if tool_executor is not None: - messages.insert( - 0, - { - "role": "system", - "content": ( - "你运行在用户明确授权的 Windows 完全访问 Agent 中。" - "凡是需要实际操作本机文件、程序、网络或桌面应用的任务，必须调用 run_powershell；" - "不要只描述计划或声称未执行的结果。命令完成后根据工具结果给出简短事实结论。" - ), - }, - ) + if toolbox is not None: request_body["tools"] = [ { "type": "function", - "function": { - "name": "run_powershell", - "description": "在当前 Agent 所在 Windows 电脑执行 PowerShell 命令，并返回标准输出或错误。", - "parameters": { - "type": "object", - "properties": {"command": {"type": "string", "description": "要执行的 PowerShell 命令。"}}, - "required": ["command"], - "additionalProperties": False, - }, - }, + "function": {"name": spec.name, "description": spec.description, "parameters": spec.parameters}, } + for spec in toolbox.specs() ] request_body["tool_choice"] = "auto" for _ in range(12): @@ -93,7 +80,7 @@ class OpenAICompatibleProvider: if not isinstance(answer, str) or not answer.strip(): raise ProviderError("LLM 响应中没有可用文本。") return visible_model_output(answer) - if tool_executor is None: + if toolbox is None: raise ProviderError("当前任务不允许模型调用本机工具。") messages.append( { @@ -110,17 +97,15 @@ class OpenAICompatibleProvider: if not isinstance(call_id, str) or not isinstance(function, dict): continue name, raw_arguments = function.get("name"), function.get("arguments") - if name != "run_powershell" or not isinstance(raw_arguments, str): + if not isinstance(name, str) or not isinstance(raw_arguments, str): result = "工具调用被拒绝：不支持的工具。" else: try: - arguments = json.loads(raw_arguments) - command = arguments.get("command") if isinstance(arguments, dict) else None - if not isinstance(command, str): - raise TypeError("命令参数无效。") - result = tool_executor(command) - except (ValueError, TypeError, json.JSONDecodeError) as error: - result = f"工具调用失败：{error}" + arguments = json.loads(raw_arguments) if raw_arguments.strip() else {} + except json.JSONDecodeError: + result = "工具调用失败：参数不是有效 JSON。" + else: + result = toolbox.call(name, arguments) messages.append({"role": "tool", "tool_call_id": call_id, "content": result}) raise ProviderError("模型连续调用本机工具次数过多，任务已安全停止。") diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py b/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py index 5442668..4265726 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py @@ -3,7 +3,6 @@ from __future__ import annotations import json -from collections.abc import Callable from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen @@ -13,18 +12,9 @@ from mulit_agent_app.infrastructure.openai_compatible_provider import ( OpenAICompatibleProvider, ProviderError, ) +from mulit_agent_app.infrastructure.tool_registry import as_toolbox, system_prompt -_FULL_ACCESS_SYSTEM = ( - "你运行在用户明确授权的 Windows 完全访问 Agent 中。" - "凡是需要实际操作本机文件、程序、网络或桌面应用的任务，必须调用 run_powershell；" - "不要只描述计划或声称未执行的结果。命令完成后根据工具结果给出简短事实结论。" -) -_FULL_ACCESS_PARAMETERS = { - "type": "object", - "properties": {"command": {"type": "string", "description": "要执行的 PowerShell 命令。"}}, - "required": ["command"], - "additionalProperties": False, -} +_MAX_TOOL_ROUNDS = 12 def build_provider(provider: str, base_url: str, model: str, credential_store: CredentialStore): @@ -46,7 +36,8 @@ class AnthropicProvider: self, instruction: str, attachments: tuple[TaskAttachment, ...] = (), - tool_executor: Callable[[str], str] | None = None, + tool_executor: object = None, + history: tuple[tuple[str, str], ...] = (), ) -> str: api_key = _require_key(self._credential_store) content: list[dict[str, object]] = [{"type": "text", "text": instruction}] @@ -54,18 +45,19 @@ class AnthropicProvider: {"type": "image", "source": {"type": "base64", "media_type": item.media_type, "data": item.base64_data}} for item in attachments ) - messages: list[dict[str, object]] = [{"role": "user", "content": content}] + # Earlier turns are plain text; Anthropic requires the first message to be + # from the user, which a stored conversation always satisfies. + messages: list[dict[str, object]] = [{"role": role, "content": text} for role, text in history] + messages.append({"role": "user", "content": content}) body: dict[str, object] = {"model": self._model, "max_tokens": 4096, "messages": messages} - if tool_executor is not None: - body["system"] = _FULL_ACCESS_SYSTEM + toolbox = as_toolbox(tool_executor) + if toolbox is not None: + body["system"] = system_prompt(toolbox) body["tools"] = [ - { - "name": "run_powershell", - "description": "在当前 Agent 所在 Windows 电脑执行 PowerShell 命令，并返回标准输出或错误。", - "input_schema": _FULL_ACCESS_PARAMETERS, - } + {"name": spec.name, "description": spec.description, "input_schema": spec.parameters} + for spec in toolbox.specs() ] - for _ in range(12): + for _ in range(_MAX_TOOL_ROUNDS): response = _anthropic_request(self._base_url, api_key, body) blocks = response.get("content") if not isinstance(blocks, list): @@ -74,19 +66,18 @@ class AnthropicProvider: if not calls: text = "\n".join(str(block.get("text", "")) for block in blocks if isinstance(block, dict)) return _text_or_error(text, "Anthropic") - if tool_executor is None: + if toolbox is None: raise ProviderError("当前任务不允许模型调用本机工具。") messages.append({"role": "assistant", "content": blocks}) results: list[dict[str, object]] = [] for call in calls: - call_id, command = call.get("id"), call.get("input", {}).get("command") if isinstance(call.get("input"), dict) else None - if not isinstance(call_id, str) or not isinstance(command, str): - result = "工具调用失败：命令参数无效。" + call_id = call.get("id") + name = call.get("name") + arguments = call.get("input") + if not isinstance(call_id, str) or not isinstance(name, str): + result = "工具调用被拒绝：不支持的工具。" else: - try: - result = tool_executor(command) - except (OSError, RuntimeError, ValueError) as error: - result = f"工具调用失败：{error}" + result = toolbox.call(name, arguments if isinstance(arguments, dict) else {}) results.append({"type": "tool_result", "tool_use_id": call_id if isinstance(call_id, str) else "invalid", "content": result}) messages.append({"role": "user", "content": results}) raise ProviderError("模型连续调用本机工具次数过多，任务已安全停止。") @@ -102,27 +93,30 @@ class GeminiProvider: self, instruction: str, attachments: tuple[TaskAttachment, ...] = (), - tool_executor: Callable[[str], str] | None = None, + tool_executor: object = None, + history: tuple[tuple[str, str], ...] = (), ) -> str: api_key = _require_key(self._credential_store) parts: list[dict[str, object]] = [{"text": instruction}] parts.extend({"inline_data": {"mime_type": item.media_type, "data": item.base64_data}} for item in attachments) - contents: list[dict[str, object]] = [{"role": "user", "parts": parts}] + # Gemini names the assistant role "model"; earlier turns are plain text. + contents: list[dict[str, object]] = [ + {"role": "user" if role == "user" else "model", "parts": [{"text": text}]} for role, text in history + ] + contents.append({"role": "user", "parts": parts}) body: dict[str, object] = {"contents": contents} - if tool_executor is not None: - body["systemInstruction"] = {"parts": [{"text": _FULL_ACCESS_SYSTEM}]} + toolbox = as_toolbox(tool_executor) + if toolbox is not None: + body["systemInstruction"] = {"parts": [{"text": system_prompt(toolbox)}]} body["tools"] = [ { "functionDeclarations": [ - { - "name": "run_powershell", - "description": "在当前 Agent 所在 Windows 电脑执行 PowerShell 命令，并返回标准输出或错误。", - "parameters": _FULL_ACCESS_PARAMETERS, - } + {"name": spec.name, "description": spec.description, "parameters": spec.parameters} + for spec in toolbox.specs() ] } ] - for _ in range(12): + for _ in range(_MAX_TOOL_ROUNDS): response = _gemini_request(self._base_url, self._model, api_key, body) try: response_parts = response["candidates"][0]["content"]["parts"] @@ -134,21 +128,17 @@ class GeminiProvider: if not calls: text = "\n".join(str(part.get("text", "")) for part in response_parts if isinstance(part, dict)) return _text_or_error(text, "Gemini") - if tool_executor is None: + if toolbox is None: raise ProviderError("当前任务不允许模型调用本机工具。") contents.append({"role": "model", "parts": response_parts}) result_parts: list[dict[str, object]] = [] for call in calls: name = call.get("name") if isinstance(call, dict) else None - args = call.get("args") if isinstance(call, dict) else None - command = args.get("command") if isinstance(args, dict) else None - if name != "run_powershell" or not isinstance(command, str): - result = "工具调用失败：命令参数无效。" + arguments = call.get("args") if isinstance(call, dict) else None + if not isinstance(name, str): + result = "工具调用被拒绝：不支持的工具。" else: - try: - result = tool_executor(command) - except (OSError, RuntimeError, ValueError) as error: - result = f"工具调用失败：{error}" + result = toolbox.call(name, arguments if isinstance(arguments, dict) else {}) result_parts.append({"functionResponse": {"name": name or "invalid", "response": {"result": result}}}) contents.append({"role": "user", "parts": result_parts}) raise ProviderError("模型连续调用本机工具次数过多，任务已安全停止。") diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py b/mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py new file mode 100644 index 0000000..a7e81c7 --- /dev/null +++ b/mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py @@ -0,0 +1,241 @@ +"""Model-facing tool set for one Agent task. + +A task's model may call the built-in full-access command tool plus every tool of +the MCP endpoints the controller deployed to this Agent. Providers only see this +registry, so adding a tool never changes provider code. +""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from dataclasses import dataclass +from functools import partial + +from mulit_agent_app.infrastructure.builtin_tools import builtin_tools +from mulit_agent_app.infrastructure.full_access_tool import FullAccessTool +from mulit_agent_app.infrastructure.mcp_client import McpClient + +RUN_POWERSHELL = "run_powershell" + +_FULL_ACCESS_PROMPT = ( + "你运行在用户明确授权的 Windows 完全访问 Agent 中。" + "凡是需要实际操作本机文件、程序、网络或桌面应用的任务，必须调用 run_powershell；" + "不要只描述计划或声称未执行的结果。命令完成后根据工具结果给出简短事实结论。" +) +_COMPUTER_PROMPT = ( + "你运行在用户明确授权的 Windows Agent 中。" + "本机文件、网址打开、时间、进程、截图和剪贴板等操作优先使用 computer_* 预设工具；" + "它们做不到或需要组合操作时，使用 run_powershell；其余能力使用工具清单中提供的 MCP 工具。" + "不要只描述计划或声称未执行的结果，调用工具后根据返回结果给出简短事实结论。" +) + +_FULL_ACCESS_PARAMETERS = { + "type": "object", + "properties": {"command": {"type": "string", "description": "要执行的 PowerShell 命令。"}}, + "required": ["command"], + "additionalProperties": False, +} + + +@dataclass(frozen=True) +class RegisteredTool: + """One tool definition a provider can render in its own wire format.""" + + name: str + description: str + parameters: dict + source: str + + +def model_tool_name(tool_id: str, remote_name: str) -> str: + """Keep MCP tool names valid for every supported provider.""" + cleaned = "".join(char if char.isalnum() or char in {"_", "-"} else "_" for char in remote_name) + prefix = "".join(char if char.isalnum() or char in {"_", "-"} else "_" for char in tool_id) + return f"{prefix}__{cleaned}"[:64] + + +def _summarise_arguments(arguments: dict) -> str: + compact = json.dumps(arguments, ensure_ascii=False, separators=(",", ":")) + return compact if len(compact) <= 200 else compact[:200] + "…" + + +def _summarise_result(result: str) -> str: + compact = " ".join(result.split()) + return compact if len(compact) <= 300 else compact[:300] + "…" + + +class McpToolBridge: + """One deployed MCP endpoint and the tools it exposes.""" + + def __init__(self, tool_id: str, client: McpClient, remote_tools: list[dict]) -> None: + self.tool_id = tool_id + self._client = client + self._remote_names: dict[str, str] = {} + tools: list[RegisteredTool] = [] + for tool in remote_tools: + remote_name = str(tool["name"]) + name = model_tool_name(tool_id, remote_name) + if name in self._remote_names: + continue + self._remote_names[name] = remote_name + schema = tool.get("inputSchema") + tools.append( + RegisteredTool( + name=name, + description=str(tool.get("description") or f"{tool_id}：{remote_name}"), + parameters=schema if isinstance(schema, dict) else {"type": "object", "properties": {}}, + source=f"mcp:{tool_id}", + ) + ) + self._tools = tuple(tools) + + def tools(self) -> tuple[RegisteredTool, ...]: + return self._tools + + def call(self, name: str, arguments: dict) -> str: + remote_name = self._remote_names.get(name) + if remote_name is None: + raise RuntimeError(f"未知的 MCP 工具：{name}") + return self._client.call_tool(remote_name, arguments) + + +class McpEndpoint: + """One deployed MCP tool: its live connection plus its model-facing tools.""" + + def __init__(self, tool_id: str, definition_key: str, client: McpClient, remote_tools: list[dict]) -> None: + self.tool_id = tool_id + self._definition_key = definition_key + self._client = client + self.bridge = McpToolBridge(tool_id, client, remote_tools) + + def matches(self, definition_key: str) -> bool: + """True when an unchanged definition can keep the running connection.""" + return definition_key == self._definition_key + + def close(self) -> None: + self._client.close() + + +class Toolbox: + """Tools available to one task, with the usage flags the executor reports.""" + + def __init__( + self, + task_id: str, + full_access: FullAccessTool | None = None, + bridges: tuple[McpToolBridge, ...] = (), + with_builtin: bool = True, + on_progress: Callable[[str], None] | None = None, + ) -> None: + self._task_id = task_id + self._full_access = full_access + self._on_progress = on_progress + self.used_full_access = False + self.used_mcp = False + self.used_builtin = False + self._tools: dict[str, RegisteredTool] = { + RUN_POWERSHELL: RegisteredTool( + name=RUN_POWERSHELL, + description="在当前 Agent 所在 Windows 电脑执行 PowerShell 命令，并返回标准输出或错误。", + parameters=_FULL_ACCESS_PARAMETERS, + source="full-access", + ) + } + self._handlers: dict[str, Callable[[dict], str]] = {RUN_POWERSHELL: self._run_powershell} + if with_builtin: + # Preset computer tools ship with the Agent, so a freshly installed + # Agent can already operate its own machine without any deployment. + for spec in builtin_tools(): + self._tools[spec.name] = RegisteredTool( + name=spec.name, + description=spec.description, + parameters=spec.parameters, + source="builtin", + ) + self._handlers[spec.name] = spec.handler + for bridge in bridges: + for tool in bridge.tools(): + self._tools[tool.name] = tool + self._handlers[tool.name] = partial(self._call_bridge, bridge, tool.name) + + def specs(self) -> tuple[RegisteredTool, ...]: + return tuple(self._tools.values()) + + @property + def used_any_tool(self) -> bool: + return self.used_full_access or self.used_mcp or self.used_builtin + + def call(self, name: str, arguments: object) -> str: + """Dispatch one model tool call. Tool failures never take down the task.""" + tool = self._tools.get(name) + handler = self._handlers.get(name) + if tool is None or handler is None: + return "工具调用被拒绝：不支持的工具。" + if not isinstance(arguments, dict): + return "工具调用失败：参数无效。" + if tool.source == "builtin": + self.used_builtin = True + self._report(f"调用 {name}：{_summarise_arguments(arguments)}") + try: + result = handler(arguments) + except (OSError, RuntimeError, TypeError, ValueError) as error: + result = f"工具调用失败：{error}" + self._report(f"{name} 返回：{_summarise_result(result)}") + return result + + def _report(self, message: str) -> None: + """Stream one step to the controller. Reporting must never break a task.""" + if self._on_progress is None: + return + try: + self._on_progress(message[:500]) + except Exception: # noqa: BLE001 - a transport hiccup cannot fail the task. + return + + def __call__(self, command: str) -> str: + """Legacy single-command seam; existing callers pass one command string.""" + return self._run_powershell({"command": command}) + + def _run_powershell(self, arguments: dict) -> str: + command = arguments.get("command") + if not isinstance(command, str): + raise TypeError("命令参数无效。") + if self._full_access is None: + raise RuntimeError("本机执行器不可用。") + self.used_full_access = True + return self._full_access.execute(self._task_id, command) + + def _call_bridge(self, bridge: McpToolBridge, name: str, arguments: dict) -> str: + self.used_mcp = True + return bridge.call(name, arguments) + + +class LegacyToolbox(Toolbox): + """Adapts a caller-supplied command runner without changing its behaviour.""" + + def __init__(self, command_runner: Callable[[str], str]) -> None: + super().__init__("", with_builtin=False) + self._command_runner = command_runner + self._handlers[RUN_POWERSHELL] = self._run_command + + def _run_command(self, arguments: dict) -> str: + command = arguments.get("command") + if not isinstance(command, str): + raise TypeError("命令参数无效。") + self.used_full_access = True + return self._command_runner(command) + + +def as_toolbox(candidate: object) -> Toolbox | None: + """Accept either a Toolbox or the previous single-command runner seam.""" + if isinstance(candidate, Toolbox): + return candidate + if callable(candidate): + return LegacyToolbox(candidate) + return None + + +def system_prompt(toolbox: Toolbox) -> str: + """Explain the available tools; a toolbox with dedicated tools needs guidance.""" + return _FULL_ACCESS_PROMPT if len(toolbox.specs()) == 1 else _COMPUTER_PROMPT diff --git a/mulit_agent/src/mulit_agent_app/main.py b/mulit_agent/src/mulit_agent_app/main.py index 5bd47d8..bae90fe 100644 --- a/mulit_agent/src/mulit_agent_app/main.py +++ b/mulit_agent/src/mulit_agent_app/main.py @@ -21,7 +21,7 @@ def main() -> int: window.root.after(0, window.hide) def quit_application() -> None: - service.session.close() + service.close() tray.stop() window.root.after(0, window.root.destroy) @@ -38,6 +38,6 @@ def main() -> int: tray.show_waiting() service.reconnect() window.run() - service.session.close() + service.close() tray.stop() return 0 diff --git a/mulit_agent/tests/conftest.py b/mulit_agent/tests/conftest.py new file mode 100644 index 0000000..93e60ad --- /dev/null +++ b/mulit_agent/tests/conftest.py @@ -0,0 +1,47 @@ +"""Shared fixtures for the Agent test suite.""" + +import sys + +import pytest + +# A minimal MCP stdio server that also pollutes stdout on purpose: a real server +# must keep working even when something logs to stdout. +FAKE_MCP_SERVER = ''' +import json +import sys + + +def reply(request_id, result): + sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": request_id, "result": result}) + "\\n") + sys.stdout.flush() + + +print("这行日志会在 stdout 上，客户端必须忽略它而不是把它当成协议数据") + +for line in sys.stdin: + line = line.strip() + if not line: + continue + message = json.loads(line) + method = message.get("method") + if method == "initialize": + reply(message["id"], {"protocolVersion": "2025-03-26"}) + elif method == "tools/list": + reply(message["id"], {"tools": [{"name": "echo", "description": "回显文本", "inputSchema": {"type": "object", "properties": {"text": {"type": "string"}}}}]}) + elif method == "tools/call": + arguments = message.get("params", {}).get("arguments") or {} + reply(message["id"], {"content": [{"type": "text", "text": "回显：" + str(arguments.get("text", ""))}]}) +''' + + +@pytest.fixture +def mcp_server_script(tmp_path): + """Return the path to a runnable fake MCP stdio server.""" + script = tmp_path / "fake_mcp_server.py" + script.write_text(FAKE_MCP_SERVER, encoding="utf-8") + return str(script) + + +@pytest.fixture +def python_executable(): + return sys.executable diff --git a/mulit_agent/tests/test_agent_service_mcp.py b/mulit_agent/tests/test_agent_service_mcp.py index 535385f..75e21fb 100644 --- a/mulit_agent/tests/test_agent_service_mcp.py +++ b/mulit_agent/tests/test_agent_service_mcp.py @@ -1,3 +1,7 @@ +import json +import time +from threading import Lock + from mulit_agent_app.application.agent_service import AgentService from mulit_agent_app.config import AgentConfig from mulit_agent_app.infrastructure.database import AgentDatabase @@ -14,12 +18,22 @@ class _Session: class _Executor: def __init__(self) -> None: self.desktop_enabled = False + self.endpoints: tuple = () def set_desktop_filesystem_enabled(self, enabled: bool) -> None: self.desktop_enabled = enabled + def set_mcp_endpoints(self, endpoints: tuple) -> None: + for endpoint in self.endpoints: + if endpoint not in endpoints: + endpoint.close() + self.endpoints = tuple(endpoints) -def test_mcp_snapshot_persists_version_and_reacknowledges_after_reconnect(tmp_path): + def mcp_endpoints(self) -> tuple: + return self.endpoints + + +def _service(tmp_path) -> AgentService: config = AgentConfig(tmp_path, tmp_path / "settings.json", tmp_path / "agent.sqlite3", tmp_path / "agent.log") service = AgentService.__new__(AgentService) service._config = config @@ -27,22 +41,36 @@ def test_mcp_snapshot_persists_version_and_reacknowledges_after_reconnect(tmp_pa service._database.initialize() service.session = _Session() service.task_executor = _Executor() - snapshot = { + service._mcp_lock = Lock() + service._mcp_sequence = 0 + return service + + +def _desktop_snapshot(version: int) -> dict: + return { "tools": [ { "id": "desktop-filesystem", - "deployment_version": 3, + "name": "桌面文件工具", + "kind": "builtin", + "version": "1.0.0", + "deployment_version": version, "enabled": True, "scope": "desktop-directories", } ] } + +def test_mcp_snapshot_persists_a_builtin_tool_and_reacknowledges_after_reconnect(tmp_path): + service = _service(tmp_path) + snapshot = _desktop_snapshot(3) + service.apply_mcp_snapshot(snapshot) service.apply_mcp_snapshot(snapshot) - assert service._database.get_meta("mcp.desktop-filesystem.enabled") == "true" - assert service._database.get_meta("mcp.desktop-filesystem.version") == "3" + stored = json.loads(service._database.get_meta("mcp.tool.desktop-filesystem")) + assert stored["deployment_version"] == 3 and stored["kind"] == "builtin" assert service.task_executor.desktop_enabled is True assert service.session.acks == [ ("desktop-filesystem", 3, "applied", ""), @@ -50,21 +78,133 @@ def test_mcp_snapshot_persists_version_and_reacknowledges_after_reconnect(tmp_pa ] -def test_mcp_snapshot_rejects_lower_version_without_changing_enabled_state(tmp_path): - config = AgentConfig(tmp_path, tmp_path / "settings.json", tmp_path / "agent.sqlite3", tmp_path / "agent.log") - service = AgentService.__new__(AgentService) - service._config = config - service._database = AgentDatabase(config.database_path) - service._database.initialize() - service._database.set_meta("mcp.desktop-filesystem.enabled", "true") - service._database.set_meta("mcp.desktop-filesystem.version", "4") - service.session = _Session() - service.task_executor = _Executor() +def test_mcp_snapshot_rejects_a_lower_version_without_changing_state(tmp_path): + service = _service(tmp_path) + service.apply_mcp_snapshot(_desktop_snapshot(4)) + service.session.acks.clear() + + service.apply_mcp_snapshot(_desktop_snapshot(3)) + + assert json.loads(service._database.get_meta("mcp.tool.desktop-filesystem"))["deployment_version"] == 4 + # The rejected snapshot is answered, and the already-deployed tool is re-acknowledged. + assert service.session.acks == [ + ("desktop-filesystem", 3, "failed", "拒绝低版本工具部署。"), + ("desktop-filesystem", 4, "applied", ""), + ] + + +def test_mcp_snapshot_rejects_an_unsupported_builtin_and_removes_unselected_tools(tmp_path): + service = _service(tmp_path) + service.apply_mcp_snapshot(_desktop_snapshot(1)) + service.session.acks.clear() service.apply_mcp_snapshot( - {"tools": [{"id": "desktop-filesystem", "deployment_version": 3, "enabled": True, "scope": "desktop-directories"}]} + { + "tools": [ + {"id": "docker", "kind": "builtin", "deployment_version": 1, "enabled": True}, + ] + } ) + assert service.session.acks == [("docker", 1, "failed", "Agent 不支持该内置工具。")] - assert service._database.get_meta("mcp.desktop-filesystem.version") == "4" + service.session.acks.clear() + service.apply_mcp_snapshot({"tools": []}) + assert service._database.get_meta("mcp.tool.desktop-filesystem") == "" assert service.task_executor.desktop_enabled is False - assert service.session.acks == [("desktop-filesystem", 3, "failed", "拒绝低版本桌面文件工具部署。")] + + +def test_http_mcp_tool_is_reported_failed_when_its_endpoint_is_unreachable(tmp_path): + service = _service(tmp_path) + + service.apply_mcp_snapshot( + { + "tools": [ + { + "id": "imageforge", + "name": "ImageForge", + "kind": "http", + "endpoint": "http://127.0.0.1:6/mcp", + "version": "1.0.0", + "deployment_version": 2, + "enabled": True, + } + ] + } + ) + deadline = time.time() + 5 + while not service.session.acks and time.time() < deadline: + time.sleep(0.05) + + assert len(service.session.acks) == 1 + tool_id, version, status, error = service.session.acks[0] + assert (tool_id, version, status) == ("imageforge", 2, "failed") and error + # A failed endpoint must not be published as a usable tool. + assert service.task_executor.endpoints == () + + +def _stdio_snapshot(command: str, args: list[str], version: int = 1) -> dict: + return { + "tools": [ + { + "id": "echo", + "name": "回显工具", + "kind": "stdio", + "command": command, + "args": args, + "version": "1.0.0", + "deployment_version": version, + "enabled": True, + } + ] + } + + +def test_stdio_tool_is_spawned_discovered_and_reused_across_reconnects(tmp_path, python_executable, mcp_server_script): + service = _service(tmp_path) + snapshot = _stdio_snapshot(python_executable, [mcp_server_script]) + + try: + service.apply_mcp_snapshot(snapshot) + _wait_for_acks(service, 1) + assert service.session.acks == [("echo", 1, "applied", "")] + endpoints = service.task_executor.endpoints + assert [endpoint.tool_id for endpoint in endpoints] == ["echo"] + process = endpoints[0]._client._transport._process + + # A reconnect replays the same definition: it must keep the live process + # instead of spawning a second server. + service.session.acks.clear() + service.apply_mcp_snapshot({**_stdio_snapshot(python_executable, [mcp_server_script], version=2)}) + _wait_for_acks(service, 1) + + assert service.session.acks == [("echo", 2, "applied", "")] + assert service.task_executor.endpoints[0] is endpoints[0] + assert process.poll() is None, "未变更的定义不应重启 MCP 进程" + finally: + service.task_executor.set_mcp_endpoints(()) + assert process.poll() is not None, "Agent 必须回收自己拉起的 MCP 进程" + + +def test_stdio_tool_is_reported_failed_when_its_command_is_missing(tmp_path): + service = _service(tmp_path) + + service.apply_mcp_snapshot(_stdio_snapshot("definitely-not-installed-command", [])) + _wait_for_acks(service, 1) + + tool_id, version, status, error = service.session.acks[0] + assert (tool_id, version, status) == ("echo", 1, "failed") and "找不到要执行的命令" in error + assert service.task_executor.endpoints == () + + +def test_stdio_tool_rejects_a_definition_with_invalid_arguments(tmp_path): + service = _service(tmp_path) + + service.apply_mcp_snapshot(_stdio_snapshot("python", ["bad\nargument"])) + + assert service.session.acks == [("echo", 1, "failed", "本机命令参数无效。")] + + +def _wait_for_acks(service: AgentService, count: int, timeout: float = 10.0) -> None: + deadline = time.time() + timeout + while len(service.session.acks) < count and time.time() < deadline: + time.sleep(0.05) diff --git a/mulit_agent/tests/test_builtin_tools.py b/mulit_agent/tests/test_builtin_tools.py new file mode 100644 index 0000000..7e3eda2 --- /dev/null +++ b/mulit_agent/tests/test_builtin_tools.py @@ -0,0 +1,149 @@ +"""The preset computer tools must work on a freshly installed Agent.""" + +from datetime import UTC, datetime, timedelta, timezone +from threading import Event +from zoneinfo import ZoneInfo + +import pytest + +from mulit_agent_app.application.task_executor import TaskExecutor +from mulit_agent_app.infrastructure import builtin_tools +from mulit_agent_app.infrastructure.builtin_tools import builtin_tools as specs +from mulit_agent_app.infrastructure.database import AgentDatabase +from mulit_agent_app.infrastructure.tool_registry import Toolbox + +_HANDLERS = {spec.name: spec.handler for spec in specs()} + + +def test_preset_tool_set_covers_file_open_time_process_screenshot_and_clipboard(): + names = set(_HANDLERS) + assert { + "computer_list_directory", + "computer_read_text_file", + "computer_write_text_file", + "computer_create_directory", + "computer_move_path", + "computer_delete_path", + "computer_open", + "computer_time", + "computer_list_processes", + "computer_screenshot", + "computer_read_clipboard", + "computer_write_clipboard", + } <= names + for spec in specs(): + assert spec.description and spec.parameters["type"] == "object" + + +def test_file_tools_round_trip_inside_a_folder(tmp_path): + folder = tmp_path / "work" + assert "已创建文件夹" in _HANDLERS["computer_create_directory"]({"path": str(folder)}) + assert "已写入" in _HANDLERS["computer_write_text_file"]({"path": str(folder / "note.txt"), "content": "第一行\n第二行"}) + + assert "第一行" in _HANDLERS["computer_read_text_file"]({"path": str(folder / "note.txt")}) + listing = _HANDLERS["computer_list_directory"]({"path": str(folder)}) + assert "note.txt" in listing and "[文件]" in listing + + moved = folder / "moved.txt" + assert "已移动" in _HANDLERS["computer_move_path"]({"source": str(folder / "note.txt"), "destination": str(moved)}) + assert _HANDLERS["computer_delete_path"]({"path": str(moved)}).startswith("已删除") + + +def test_file_tools_refuse_dangerous_or_accidental_overwrites(tmp_path): + existing = tmp_path / "keep.txt" + existing.write_text("原内容", encoding="utf-8") + + with pytest.raises(ValueError, match="overwrite=true"): + _HANDLERS["computer_write_text_file"]({"path": str(existing), "content": "新内容"}) + assert existing.read_text(encoding="utf-8") == "原内容" + _HANDLERS["computer_write_text_file"]({"path": str(existing), "content": "新内容", "overwrite": True}) + assert existing.read_text(encoding="utf-8") == "新内容" + + with pytest.raises(ValueError, match="不是可读取的文件"): + _HANDLERS["computer_read_text_file"]({"path": str(tmp_path / "missing.txt")}) + + folder = tmp_path / "nested" + (folder / "child").mkdir(parents=True) + (folder / "child" / "file.txt").write_text("x", encoding="utf-8") + with pytest.raises(ValueError, match="recursive=true"): + _HANDLERS["computer_delete_path"]({"path": str(folder)}) + assert _HANDLERS["computer_delete_path"]({"path": str(folder), "recursive": True}).startswith("已删除") + + with pytest.raises(ValueError, match="拒绝删除"): + _HANDLERS["computer_delete_path"]({"path": "C:\\Windows", "recursive": True}) + + +def test_time_tool_accepts_local_offsets_and_iana_zones(): + assert "星期" in _HANDLERS["computer_time"]({}) + + for requested in ("Asia/Tokyo", "UTC", "UTC+9"): + before = datetime.now(_zone(requested)).strftime("%Y-%m-%d %H") + reported = _HANDLERS["computer_time"]({"timezone": requested}) + after = datetime.now(_zone(requested)).strftime("%Y-%m-%d %H") + assert before <= reported[:13] <= after, f"{requested} 返回了错误时间：{reported}" + + with pytest.raises(ValueError, match="无法识别时区"): + _HANDLERS["computer_time"]({"timezone": "Mars/Olympus"}) + + +def _zone(requested: str): + if requested == "UTC": + return UTC + if requested == "UTC+9": + return timezone(timedelta(hours=9)) + return ZoneInfo(requested) + + +def test_open_tool_opens_urls_and_refuses_unknown_paths(monkeypatch): + opened: list[str] = [] + monkeypatch.setattr(builtin_tools.webbrowser, "open", lambda target: opened.append(target) or True) + + assert "已用系统默认程序打开网址" in _HANDLERS["computer_open"]({"target": "https://example.com"}) + assert opened == ["https://example.com"] + + with pytest.raises(ValueError, match="只支持 http/https"): + _HANDLERS["computer_open"]({"target": "ftp://example.com"}) + + +def test_process_screenshot_and_clipboard_use_the_local_windows_session(tmp_path): + processes = _HANDLERS["computer_list_processes"]({"limit": 5}) + assert "Name" in processes or "名称" in processes + + shot = _HANDLERS["computer_screenshot"]({"path": str(tmp_path / "shot.png")}) + assert "已保存屏幕截图" in shot + assert (tmp_path / "shot.png").read_bytes()[:8] == b"\x89PNG\r\n\x1a\n" + + assert "已写入剪贴板" in _HANDLERS["computer_write_clipboard"]({"text": "MultiClaw 剪贴板测试"}) + assert "MultiClaw 剪贴板测试" in _HANDLERS["computer_read_clipboard"]({}) + + +def test_toolbox_exposes_preset_tools_and_reports_builtin_execution(tmp_path): + toolbox = Toolbox("task-1", None) + names = {spec.name for spec in toolbox.specs()} + assert "run_powershell" in names and "computer_time" in names + + assert "星期" in toolbox.call("computer_time", {"timezone": "UTC"}) + assert toolbox.used_builtin is True and toolbox.used_any_tool is True + assert toolbox.call("computer_unknown", {}).startswith("工具调用被拒绝") + assert toolbox.call("computer_read_text_file", {"path": "C:\\definitely\\missing.txt"}).startswith("工具调用失败") + + +def test_task_executor_marks_preset_tool_usage(tmp_path): + database = AgentDatabase(tmp_path / "agent.sqlite3") + + def runner(_instruction, _attachments, toolbox): + return f"本机时间：{toolbox.call('computer_time', {'timezone': 'UTC'})}" + + executor = TaskExecutor(database, runner) + 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("preset-task", "看看现在几点") + assert completed.wait(10) + assert events[-1]["execution"] == "builtin" + assert "本机时间：" in str(events[-1]["output"]) diff --git a/mulit_agent/tests/test_controller_session.py b/mulit_agent/tests/test_controller_session.py index ed86d79..83c88e2 100644 --- a/mulit_agent/tests/test_controller_session.py +++ b/mulit_agent/tests/test_controller_session.py @@ -44,3 +44,33 @@ def test_replaced_socket_close_does_not_schedule_an_extra_reconnect(tmp_path): session._on_close(object(), None, None) # type: ignore[arg-type] assert scheduled == [] + + +def test_conversation_history_is_validated_and_kept_in_order(): + history = ControllerSession._validated_history( + [ + {"role": "user", "content": "下载到 D:\\demo"}, + {"role": "assistant", "content": "已下载到 D:\\demo\\a.zip"}, + {"role": "system", "content": "不该被接受"}, + {"role": "assistant", "content": " "}, + "not-a-mapping", + ] + ) + + assert history == (("user", "下载到 D:\\demo"), ("assistant", "已下载到 D:\\demo\\a.zip")) + + +def test_conversation_history_is_bounded_by_budget_and_drops_the_oldest_turns(): + entries = [{"role": "user", "content": "x" * 4_000}, {"role": "assistant", "content": "y" * 4_000}] * 10 + + history = ControllerSession._validated_history(entries) + + assert history, "至少应保留最近的一轮" + assert len(history) * 4_000 <= 60_000 + assert history[-1][1].startswith("y"), "保留的应是最新的轮次" + + +def test_malformed_history_never_raises(): + assert ControllerSession._validated_history(None) == () + assert ControllerSession._validated_history({"role": "user"}) == () + assert ControllerSession._validated_history([{"role": "user", "content": 42}]) == () diff --git a/mulit_agent/tests/test_image_attachments.py b/mulit_agent/tests/test_image_attachments.py index d4daafb..6467b4e 100644 --- a/mulit_agent/tests/test_image_attachments.py +++ b/mulit_agent/tests/test_image_attachments.py @@ -50,10 +50,50 @@ def test_provider_image_payloads(monkeypatch): assert captured[2]["contents"][0]["parts"][1]["inline_data"]["mime_type"] == "image/png" +def test_anthropic_and_gemini_receive_the_conversation_history(monkeypatch): + 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())) + if request.full_url.endswith("/messages"): + return Response({"content": [{"type": "text", "text": "好的"}]}) + return Response({"candidates": [{"content": {"parts": [{"text": "好的"}]}}]}) + + monkeypatch.setattr("mulit_agent_app.infrastructure.provider_factory.urlopen", fake_urlopen) + history = (("user", "下载到 D:\\demo"), ("assistant", "已下载到 D:\\demo\\a.zip")) + + assert AnthropicProvider("https://example", "model", Credentials()).generate("把它复制到 D:\\target", history=history) == "好的" + assert GeminiProvider("https://example", "model", Credentials()).generate("把它复制到 D:\\target", history=history) == "好的" + + assert captured[0]["messages"] == [ + {"role": "user", "content": "下载到 D:\\demo"}, + {"role": "assistant", "content": "已下载到 D:\\demo\\a.zip"}, + {"role": "user", "content": [{"type": "text", "text": "把它复制到 D:\\target"}]}, + ] + assert captured[1]["contents"] == [ + {"role": "user", "parts": [{"text": "下载到 D:\\demo"}]}, + {"role": "model", "parts": [{"text": "已下载到 D:\\demo\\a.zip"}]}, + {"role": "user", "parts": [{"text": "把它复制到 D:\\target"}]}, + ] + + def test_anthropic_and_gemini_complete_local_tool_calls(monkeypatch): responses = { "anthropic": [ - {"content": [{"type": "tool_use", "id": "anthropic-call", "input": {"command": "Write-Output anth"}}]}, + {"content": [{"type": "tool_use", "id": "anthropic-call", "name": "run_powershell", "input": {"command": "Write-Output anth"}}]}, {"content": [{"type": "text", "text": "Anthropic 已执行。"}]}, ], "gemini": [ diff --git a/mulit_agent/tests/test_mcp_tools.py b/mulit_agent/tests/test_mcp_tools.py new file mode 100644 index 0000000..01be752 --- /dev/null +++ b/mulit_agent/tests/test_mcp_tools.py @@ -0,0 +1,255 @@ +"""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() == () diff --git a/mulit_agent/tests/test_openai_compatible_provider.py b/mulit_agent/tests/test_openai_compatible_provider.py index 2f2fb96..a07b942 100644 --- a/mulit_agent/tests/test_openai_compatible_provider.py +++ b/mulit_agent/tests/test_openai_compatible_provider.py @@ -42,6 +42,42 @@ def test_provider_posts_a_chat_completion(monkeypatch, tmp_path): } +def test_provider_sends_the_conversation_history_before_the_new_turn(monkeypatch): + captured = {} + + def fake_urlopen(request, timeout): + captured["body"] = json.loads(request.data.decode("utf-8")) + return _Response() + + monkeypatch.setattr("mulit_agent_app.infrastructure.openai_compatible_provider.urlopen", fake_urlopen) + + output = OpenAICompatibleProvider("https://api.example.test/v1", "example-model", _Credentials()).generate( + "把它复制到 D:\\target", + history=(("user", "下载到 D:\\demo"), ("assistant", "已下载到 D:\\demo\\a.zip")), + ) + + assert output == "模型结果" + assert captured["body"]["messages"] == [ + {"role": "user", "content": "下载到 D:\\demo"}, + {"role": "assistant", "content": "已下载到 D:\\demo\\a.zip"}, + {"role": "user", "content": "把它复制到 D:\\target"}, + ] + + +def test_provider_without_history_still_sends_a_single_user_message(monkeypatch): + captured = {} + + def fake_urlopen(request, timeout): + captured["body"] = json.loads(request.data.decode("utf-8")) + return _Response() + + monkeypatch.setattr("mulit_agent_app.infrastructure.openai_compatible_provider.urlopen", fake_urlopen) + + OpenAICompatibleProvider("https://api.example.test/v1", "example-model", _Credentials()).generate("请总结") + + assert captured["body"]["messages"] == [{"role": "user", "content": "请总结"}] + + def test_provider_removes_reasoning_tags_before_returning_result(monkeypatch): class ThinkingResponse(_Response): def read(self): diff --git a/mulit_agent/tests/test_task_executor.py b/mulit_agent/tests/test_task_executor.py index c7ee498..dfd1694 100644 --- a/mulit_agent/tests/test_task_executor.py +++ b/mulit_agent/tests/test_task_executor.py @@ -216,6 +216,80 @@ def test_executor_does_not_fall_back_to_llm_when_desktop_tool_is_not_applied(tmp assert "未部署或尚未应用" in str(events[-1]["error"]) +def test_executor_streams_tool_progress_while_the_task_runs(tmp_path): + database = AgentDatabase(tmp_path / "agent.sqlite3") + + def runner(_instruction, _attachments, toolbox): + return f"结果：{toolbox.call('computer_time', {'timezone': 'UTC'})}" + + executor = TaskExecutor(database, runner) + 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("progress-task", "看看现在几点") + + assert completed.wait(10) + progress = [str(event["message"]) for event in events if event.get("status") == "progress"] + assert any(message.startswith("调用 computer_time") for message in progress), progress + assert any(message.startswith("computer_time 返回：") for message in progress), progress + # Progress must arrive before the terminal event so the UI can stream it live. + statuses = [event.get("status") for event in events] + assert statuses.index("progress") < statuses.index("completed") + + +def test_executor_ignores_progress_after_the_task_ended(tmp_path): + database = AgentDatabase(tmp_path / "agent.sqlite3") + executor = TaskExecutor(database, lambda instruction: "完成") + 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("finished-task", "总结") + assert completed.wait(5) + reported = len(events) + + executor._report_progress("finished-task")("迟到的进展") + + assert len(events) == reported, "结束后的任务不应再产生中间步骤" + + +def test_executor_passes_the_conversation_history_to_the_provider(tmp_path): + database = AgentDatabase(tmp_path / "agent.sqlite3") + received: list[tuple[tuple[str, str], ...]] = [] + + def runner(_instruction, _attachments, _toolbox, history): + received.append(history) + return "已复制到 D:\\target" + + executor = TaskExecutor(database, runner) + completed = Event() + executor.task_event.connect( + lambda event: completed.set() if isinstance(event, dict) and event.get("status") == "completed" else None + ) + + executor.submit( + "follow-up-task", + "把它复制到 D:\\target", + history=(("user", "下载到 D:\\demo"), ("assistant", "已下载到 D:\\demo\\a.zip")), + ) + + assert completed.wait(5) + assert received == [(("user", "下载到 D:\\demo"), ("assistant", "已下载到 D:\\demo\\a.zip"))] + + def test_image_task_never_falls_back_to_text_only_model(tmp_path): database = AgentDatabase(tmp_path / "agent.sqlite3") calls: list[str] = []