aps-agent/server/gateway/agent_api.py

# ============================================================
# Pi Agent 外向接入 API（moduleId: gateway-agent-api, 可重生 ✅）
# 方向 A · P1：/api/agent/* 薄适配层。
#   - 目录派生自 tool_runtime 只读白名单 + harness _POWER_MAP P0；
#   - invoke 只直通目录内意图；P1/P2 返回 CONFIRM_REQUIRED，P3 返回 POWER_DENIED；
#   - 所有业务委托 run_tool_async / handle_intent，绝不复制或旁路写路径。
# ============================================================
from __future__ import annotations

import logging
import uuid
from typing import Any

from fastapi import APIRouter
from fastapi.responses import JSONResponse
from pydantic import BaseModel, ConfigDict, Field

from server.contracts import INTERFACE_VERSION

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/agent", tags=["pi-agent"])


# P1 agent catalog is an explicit allowlist, not an automatic projection of P0
# intents: tool_runtime/readonly naming has historically mixed in handlers that
# write the world, knowledge base, or report files (data.analyze/folder.analyze/
# report.generate). Keeping the list explicit means a future P0 registration
# does not silently become reachable by an external read-only agent token.
_AGENT_CATALOG_INTENTS = frozenset(
    {
        "help",
        "guidance.next",
        "master.query",
        "plan.buckets",
        "plan.rccp",
        "plan.feasibility",
        "plan.inventory",
        "plan.leveling",
        "plan.supply",
        "plan.trace",
        "knowledge.query",
        "scenario.compare",
        "order.pool",
        "conflict.list",
        "flex.capacity",
        "flex.compare",
        "flex.simulate_due",
        "skill.list",
        "query.kpi",
    }
)


class AgentInvokeRequest(BaseModel):
    """Pi Agent 只读意图调用请求（模块级模型，防止 OpenAPI 前向引用失败）。"""

    model_config = ConfigDict(extra="forbid")

    intent: str = Field(min_length=1, max_length=128)
    params: dict[str, Any] = Field(default_factory=dict)
    requestId: str | None = Field(default=None, max_length=128)
    sessionId: str | None = Field(default=None, max_length=128)


def _parameter_schema() -> dict[str, Any]:
    """保守意图参数槽：不臆造各意图业务字段，未知键由服务端登记语义处置。"""
    return {
        "type": "object",
        "title": "意图参数",
        "description": "参数透传槽位；多数只读查询可传空对象。服务端对未登记/越权行为 fail closed。",
        "properties": {},
        "additionalProperties": True,
    }


def _catalog_items() -> list[dict[str, Any]]:
    from server.agent_core import harness

    entries: list[dict[str, Any]] = []
    for name in sorted(_AGENT_CATALOG_INTENTS):
        power = harness.power_of(name)
        entries.append(
            {
                "name": name,
                "power": power,
                "description": harness._POLICY_DESC.get(
                    name, "Pi Agent 只读 APS 查询/分析意图"
                ),
                "readOnly": True,
                "parameters": _parameter_schema(),
            }
        )
    return entries


def _catalog_map() -> dict[str, dict[str, Any]]:
    return {item["name"]: item for item in _catalog_items()}


def _not_found(message: str) -> JSONResponse:
    return JSONResponse(
        status_code=404,
        content={
            "ok": False,
            "error": {"code": "INTENT_NOT_FOUND", "message": message},
        },
    )


def _deny(
    store: Any,
    *,
    intent: str,
    power: str,
    actor: str,
    code: str,
    message: str,
    request_id: str,
    status_code: int,
) -> JSONResponse:
    try:
        from server.agent_core.audit import write_audit

        write_audit(
            store.data,
            store.next_id,
            actor=actor,
            category="AGENT",
            action="agent.invoke.denied",
            target={"type": "INTENT", "id": intent},
            power=power,
            rationale={
                "reason": code,
                "requestId": request_id,
                "source": "pi-agent-gateway",
            },
            result="DENIED",
        )
        store.save()
    except Exception:
        logger.warning("agent invoke denied audit failed for %s", intent, exc_info=True)
    return JSONResponse(
        status_code=status_code,
        content={"ok": False, "error": {"code": code, "message": message}},
    )


@router.get("/intents", response_model=None)
def list_intents() -> dict[str, Any]:
    """返回当前只读目录，供 Pi 动态生成工具清单。"""
    return {
        "ok": True,
        "scope": ["read"],
        "intents": _catalog_items(),
        "interfaceVersion": INTERFACE_VERSION,
    }


@router.get("/intents/{name}", response_model=None)
def get_intent(name: str) -> dict[str, Any] | JSONResponse:
    """返回单个意图的参数目录项。"""
    key = str(name or "").strip()
    item = _catalog_map().get(key)
    if item is None:
        return _not_found(f"意图「{key}」不在只读目录")
    return item


@router.post("/invoke", response_model=None)
async def invoke(req: AgentInvokeRequest) -> dict[str, Any] | JSONResponse:
    """执行目录内只读意图，业务委托统一工具运行时。"""
    from server.agent_core import harness
    from server.agent_core.tool_runtime import is_registered, run_tool_async
    from server.auth.context import get_identity
    from server.contracts import IntentResult
    from server.state.store import get_store

    identity = get_identity(required=True)
    actor = (
        f"pi-agent:{identity.username}"
        if identity.auth_kind == "agent"
        else identity.username
    )
    request_id = (req.requestId or "").strip() or uuid.uuid4().hex
    name = (req.intent or "").strip()
    store = get_store()

    if not is_registered(name):
        return _deny(
            store,
            intent=name,
            power=harness.power_of(name),
            actor=actor,
            code="INTENT_NOT_REGISTERED",
            message=f"意图「{name}」未登记，拒绝执行",
            request_id=request_id,
            status_code=404,
        )

    power = harness.power_of(name)
    if name not in _catalog_map():
        if power == "P3":
            return _deny(
                store,
                intent=name,
                power=power,
                actor=actor,
                code="POWER_DENIED",
                message="该意图属于 P3 外部副作用，Pi Agent 只读接入无权执行",
                request_id=request_id,
                status_code=403,
            )
        return _deny(
            store,
            intent=name,
            power=power,
            actor=actor,
            code="CONFIRM_REQUIRED",
            message="该意图尚未开放给 Pi Agent 直通，需人工确认后才能执行",
            request_id=request_id,
            status_code=409,
        )

    session_id = (req.sessionId or "").strip() or f"agent:{actor}"
    intent = IntentResult.model_construct(
        intent=name,
        params=dict(req.params or {}),
        confidence=1.0,
        source="LLM",
    )
    try:
        reply = await run_tool_async(store, session_id, intent, actor=actor)
    except Exception as exc:
        logger.exception("agent invoke upstream failed for %s", name, exc_info=exc)
        return JSONResponse(
            status_code=502,
            content={
                "ok": False,
                "error": {
                    "code": "UPSTREAM_FAILED",
                    "message": "上游业务处理失败，本次调用未完成",
                    "requestId": request_id,
                },
            },
        )

    dumped = reply.model_dump()
    blocks = list(dumped.get("blocks") or [])
    evidence_refs: list[str] = []
    for block in blocks:
        for ref in block.get("evidenceRefs") or []:
            if ref not in evidence_refs:
                evidence_refs.append(str(ref))
    return {
        "ok": True,
        "intent": name,
        "power": power,
        "requestId": request_id,
        "actor": actor,
        "data": {
            "text": reply.text,
            "blocks": blocks,
            "commands": list(dumped.get("commands") or []),
            "evidenceRefs": evidence_refs,
        },
        "interfaceVersion": INTERFACE_VERSION,
    }