aps-agent/server/gateway/agent_api.py

268 lines
8.3 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# ============================================================
# 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,
}