"""基座验证探针:端到端触发 `ToolExecutor` 的四种拒绝分支。 **为什么专门做一个 Agent**:`ToolExecutor` 的四种拒绝(意图未配置工具白名单 / 工具不在 白名单 / 缺权限 / 角色不符)在真实链路上很难**安全**触发 —— 要么去改客服、风控的生效配置, 要么去动 RBAC,两条路都会影响正在工作的 Agent。用一个只读、无副作用的探针把这件事隔离出来。 **它不碰任何业务数据**:唯一的工具只回显调用方给的参数,不查库、不调模型、不写状态。 角色限定为 `admin`,意图只有 `probe` 一个。 用法见 `tools/verify_tool_executor_denials.py`:它按顺序调整探针的意图配置与工具白名单, 分别触发四种拒绝,最后把配置恢复到验证前的状态。 """ from typing import Any from pydantic import BaseModel, ConfigDict from app.core.contracts import ( AgentDefinition, AgentRequest, CoreResult, RequestContext, ) from app.service.agent.base import BaseAgent AGENT_TYPE = "platform_probe" INTENT_PROBE = "probe" PROBE_TOOL = "probe_echo" PROBE_PERMISSION = "probe:read" class ProbeEchoArgs(BaseModel): """严格入参:探针不接收自由文本,避免被当成通用执行入口。""" model_config = ConfigDict(extra="forbid", frozen=True) note: str = "" async def probe_echo_tool(arguments: BaseModel, context: RequestContext) -> dict[str, Any]: """回显参数。**只读且无副作用** —— 不查库、不写状态、不调外部服务。""" note = getattr(arguments, "note", "") return {"echo": note, "trace_id": context.trace_id} class PlatformProbeAgent(BaseAgent): """基座探针:只用于验证工具链路的拒绝行为,不承载任何业务。""" definition = AgentDefinition( agent_type=AGENT_TYPE, version="1.0.0", allowed_roles=("admin",), allowed_portals=("api",), allowed_tools=(PROBE_TOOL,), supported_intents=(INTENT_PROBE,), ) async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult: # 走公共工具链路:探针要验证的正是这条路上的白名单、权限与角色校验。 output = await self.call_tool( PROBE_TOOL, {"note": request.message[:50]}, intent=INTENT_PROBE, context=context, ) return CoreResult(text=f"probe ok: {output}", intent=self._classified_intent)