"""百工执行器:按能力执行类型分派(builtin / qiankun 乾坤 / mcp / http)。 乾坤经 ``integrations/sandbox`` 的 SandboxClient——HMAC 短期凭证授契由沙箱客户端 签发(SANDBOX_AUTH_SECRET 两端一致),执行审计证据经行动账落库。 """ from __future__ import annotations import json import shlex from dataclasses import dataclass from typing import Any import httpx from sqlalchemy import select from app.core.errors import AppError from app.db.session import AsyncSessionLocal from app.domains.baigong.models import Capability from app.integrations.mcp.client import call_tool from app.integrations.sandbox.client import SandboxClient from app.integrations.sandbox.models import ExecRequest as SandboxExecRequest _HTTP_TIMEOUT_SECONDS = 30.0 @dataclass class ExecOutcome: ok: bool output: str error_code: str | None = None duration_ms: int = 0 class Executor: """能力执行分派器。构造时注入乾坤客户端(未配置则乾坤类能力 503)。""" def __init__(self, sandbox: SandboxClient | None, http: httpx.AsyncClient | None = None) -> None: self._sandbox = sandbox self._http = http async def execute(self, capability: Capability, params: dict[str, Any]) -> ExecOutcome: dispatch = { "builtin": self._exec_builtin, "qiankun": self._exec_qiankun, "mcp": self._exec_mcp, "http": self._exec_http, } handler = dispatch.get(capability.executor_type) if handler is None: raise AppError("invalid_request", f"未知执行类型:{capability.executor_type}") return await handler(capability, params) # ── 内置:无副作用纯计算(时间/回显等) ── async def _exec_builtin(self, capability: Capability, params: dict[str, Any]) -> ExecOutcome: from app.core.time import to_iso, utc_now if capability.executor_ref == "echo": return ExecOutcome(ok=True, output=f"[echo] {params.get('text', '')}") if capability.executor_ref == "datetime.now": return ExecOutcome(ok=True, output=to_iso(utc_now())) raise AppError("not_found", f"内置能力不存在:{capability.executor_ref}") # ── 乾坤:不可信执行隔离(HMAC 授契由 SandboxClient 签发) ── async def _exec_qiankun(self, capability: Capability, params: dict[str, Any]) -> ExecOutcome: if self._sandbox is None: raise AppError( "service_unavailable", "乾坤沙箱未就绪(SANDBOX 未配置),无法执行。", status_code=503 ) command = str(params.get("command", "")) if not command: raise AppError("invalid_request", "qiankun 能力需要 command 参数") result = await self._sandbox.exec(SandboxExecRequest(argv=shlex.split(command))) output = result.stdout or result.stderr or "" return ExecOutcome( ok=result.ok, output=output, error_code=None if result.ok else "qiankun_exit", duration_ms=result.duration_ms, ) # ── MCP:外部能力服务(博闻等);executor_ref = mcp_servers.name ── async def _exec_mcp(self, capability: Capability, params: dict[str, Any]) -> ExecOutcome: from app.models.mcp_servers import McpServer async with AsyncSessionLocal() as session: server = ( await session.execute( select(McpServer).where(McpServer.name == capability.executor_ref) ) ).scalars().first() if server is None or not server.endpoint_url: raise AppError( "not_found", f"MCP 服务未登记或无端点:{capability.executor_ref}" ) endpoint, token = server.endpoint_url, server.auth_token session.expunge(server) client = self._http or httpx.AsyncClient(trust_env=False, timeout=_HTTP_TIMEOUT_SECONDS) result = await call_tool(client, endpoint, capability.name.removeprefix(f"mcp_{capability.executor_ref}_"), params, token) payload = getattr(result, "content", None) or result return ExecOutcome(ok=True, output=json.dumps(payload, ensure_ascii=False, default=str)) # ── HTTP:受授契约束的外部 HTTP 动作 ── async def _exec_http(self, capability: Capability, params: dict[str, Any]) -> ExecOutcome: url = str(params.get("url", "")) if not url: raise AppError("invalid_request", "http 能力需要 url 参数") async with httpx.AsyncClient(trust_env=False, timeout=_HTTP_TIMEOUT_SECONDS) as client: resp = await client.request(str(params.get("method", "GET")).upper(), url) return ExecOutcome( ok=resp.status_code < 400, output=f"HTTP {resp.status_code}: {resp.text[:500]}", error_code=None if resp.status_code < 400 else f"http_{resp.status_code}", )