hunjing/backend/app/domains/baigong/executor.py

"""百工执行器：按能力执行类型分派（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}",
        )