Compare commits

..

2 Commits

9 changed files with 872 additions and 38 deletions

View File

@ -5,6 +5,10 @@
> 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。 > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
> 工程口径的完整记录见 `PROGRESS.md` / git log。 > 工程口径的完整记录见 `PROGRESS.md` / git log。
## 0.62.1 — 2026-08-06
- 修复外部系统只读 POST 接口因请求体结构信息缺失而反复报参数错误的问题;大查询结果不再被截成半段内容,超出对话容量时会完整暂存并支持按路径、分页和字段继续读取。
## 0.62.0 — 2026-08-05 ## 0.62.0 — 2026-08-05
- 外部系统连接不再局限于 Factory MES管理员现在可以直接配置标准 OpenAPI 系统,并选择用户名密码换取 Token、API Key 或 Bearer Token用户连接页面会按系统要求动态显示凭据字段。现有 Factory 配置和账号连接无需迁移。 - 外部系统连接不再局限于 Factory MES管理员现在可以直接配置标准 OpenAPI 系统,并选择用户名密码换取 Token、API Key 或 Bearer Token用户连接页面会按系统要求动态显示凭据字段。现有 Factory 配置和账号连接无需迁移。

View File

@ -410,14 +410,36 @@ scheduled_jobs(§8.5) channel_bindings(§8.7,判别列+JSONB)
- Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。 - Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。
- Swagger/OpenAPI JSON 不持久化入数据库或文件,连接器按 `definition_id + user_id` 隔离后放在进程内存中缓存 5 分钟;重启自动失效。这样保留实时契约发现,又避免不同身份可见的规格互相污染。 - Swagger/OpenAPI JSON 不持久化入数据库或文件,连接器按 `definition_id + user_id` 隔离后放在进程内存中缓存 5 分钟;重启自动失效。这样保留实时契约发现,又避免不同身份可见的规格互相污染。
**工具面**:不把数百个 Swagger operation 全展开为 JSON tool(工具列表膨胀+选择降准),只挂个 host-side 元工具:`external_system_list`(已连系统 + 管理员查询规划提示),`external_system_search`(按问题搜 operation 摘要 + 置顶管理员推荐入口),`external_system_call`(按 operation_id 调用)。仅当该 user 有 active 连接时注册,密钥不进 sandbox。搜索只展示实际可调用的 GET/HEAD 和已放行 POST管理员在 definition JSONB 配置 `query_guidance``recommended_operation_ids`,前者是可信控制面的软路由策略,后者是无需关键词命中的机械发现入口。Factory 默认把 BI dataset list/exec 作为统计聚合入口,日志/明细用于逐条追溯Swagger 业务文本仍是不可信数据。 **工具面**:不把数百个 Swagger operation 全展开为 JSON tool(工具列表膨胀+选择降准),只挂个 host-side 元工具:`external_system_list`(已连系统 + 管理员查询规划提示),`external_system_search`(按问题搜 operation 摘要、解析后的请求 body schema + 置顶管理员推荐入口),`external_system_call`(按 operation_id 调用),`external_system_result_read`(按 `result_ref` + JSON Pointer/分页/字段投影读取大响应)。仅当该 user 有 active 连接时注册,密钥不进 sandbox。搜索只展示实际可调用的 GET/HEAD 和已放行 POST管理员在 definition JSONB 配置 `query_guidance``recommended_operation_ids`,前者是可信控制面的软路由策略,后者是无需关键词命中的机械发现入口。Factory 默认把 BI dataset list/exec 作为统计聚合入口,日志/明细用于逐条追溯Swagger 业务文本仍是不可信数据。
**明细扫描边界**:单次响应保留字节上限,每次 agent run 另按外部系统累计返回量Factory connector 将 `page_size` 限在管理员上限,拒绝 `page=0` / `pageoff` 关闭分页。三者防模型通过连续翻日志自行做昂贵聚合,但不改变 Factory 对其他客户端的分页契约。达到边界后工具正向引导回 dataset/聚合接口或缩小查询范围。 **大响应**:`max_result_bytes` 是进入模型上下文的单次内联额度,不再用于切断原始 JSON超额响应完整写入当前 user_root 下按 task_id 隔离的隐藏缓存,工具只返回合法结构化预览、`result_ref`、原始字节数和可继续读取的位置。reader 每次读取都重新校验当前 user 对原 external system 的 active 授权,并与 call 共享本轮 `max_total_result_bytes` 内联额度。缓存固定 24h TTL、单响应 10 MiB、单 task 50 MiB、单 user 200 MiB,过期或超额时优先清理最旧缓存;超过响应安全上限的远端结果直接拒绝并要求缩小范围,不产生半截 JSON。这里把“上游响应安全边界”“完整结果保存”“模型上下文额度”拆成三层,既不丢数据,也不靠无限提高上下文额度解决大结果问题。
**明细扫描边界**:单次响应保留安全上限与模型内联额度,每次 agent run 另按外部系统累计内联返回量Factory connector 将 `page_size` 限在管理员上限,拒绝 `page=0` / `pageoff` 关闭分页。三者防模型通过连续翻日志自行做昂贵聚合,但不改变 Factory 对其他客户端的分页契约。达到边界后工具正向引导回 dataset/聚合接口、`result_ref` 分段读取或缩小查询范围。
**状态与 UI两表**:`external_system_definitions` 保存管理员维护的可信系统目录、查询规划提示、推荐入口和 `access_mode=all|selected`;这些新增项复用既有 `config` JSONB,无 schema/migration。提示词在 admin 表单里复用通用 dialog 的多行编辑器,不把长文常驻铺在页面。`external_systems` 同时承载指定用户授权和用户密文连接,`pending` 表示已授权但未配置凭据,`active` 才挂工具。管理员撤销指定用户会删除其连接和密文凭据;用户自行断开只清凭据、保留管理员授权。管理后台可新增、编辑、停用目录项,已有用户连接的目录项禁止直接删除。左栏「外部系统」面板只能选择当前用户可见目录、测试连接、替换凭据和断开,不能查看密码。稳定问法沉淀到用户私有 skill 时只写 provider/operation_id/参数规则,永远使用当前提问者的连接执行,共享 skill 不等于共享权限。 **状态与 UI两表**:`external_system_definitions` 保存管理员维护的可信系统目录、查询规划提示、推荐入口和 `access_mode=all|selected`;这些新增项复用既有 `config` JSONB,无 schema/migration。提示词在 admin 表单里复用通用 dialog 的多行编辑器,不把长文常驻铺在页面。`external_systems` 同时承载指定用户授权和用户密文连接,`pending` 表示已授权但未配置凭据,`active` 才挂工具。管理员撤销指定用户会删除其连接和密文凭据;用户自行断开只清凭据、保留管理员授权。管理后台可新增、编辑、停用目录项,已有用户连接的目录项禁止直接删除。左栏「外部系统」面板只能选择当前用户可见目录、测试连接、替换凭据和断开,不能查看密码。稳定问法沉淀到用户私有 skill 时只写 provider/operation_id/参数规则,永远使用当前提问者的连接执行,共享 skill 不等于共享权限。
**不选**:①zcbot 直连 Factory DB(绕过现有 RBAC/审计,只读仍可越权/拖垮主库);②固定几个查询模板(把 agent 降成菜单,无法利用 Factory 已有广泛 API);③直接复用 Factory `ichat` 自由 SQL 原型(字符串安全判断不构成边界,且使用默认 DB 凭据);④自动把相似问题生成并上线新代码工具(候选配方可自动生成,可执行能力仍需工具门控/人审)。 **不选**:①zcbot 直连 Factory DB(绕过现有 RBAC/审计,只读仍可越权/拖垮主库);②固定几个查询模板(把 agent 降成菜单,无法利用 Factory 已有广泛 API);③直接复用 Factory `ichat` 自由 SQL 原型(字符串安全判断不构成边界,且使用默认 DB 凭据);④自动把相似问题生成并上线新代码工具(候选配方可自动生成,可执行能力仍需工具门控/人审)。
### 8.15 OpenWorker 对照:动作治理与可恢复的人类介入(design,2026-08-06)
**调研结论**:OpenWorker 与 zcbot 同样采用单 agent loop、模型自由、文件产物、skill 渐进披露、持久会话、定时任务和外部连接器;这些不是 zcbot 的新增方向。真正值得借鉴的是它把「agent 能做什么」与「人此刻是否在线」拆成两条正交轴:统一权限引擎先按动作风险决定 allow/deny/ask,审批、问题、通知再进入同一个可持久化 attention inbox人在 Web 或消息渠道任一端回应后,原调用幂等恢复。对 zcbot 的增量因此不是再造 agent/工作流框架,而是在现有 loop、scheduler、channel、external systems 之上补一层**动作治理控制面**。参考实现以 `andrewyng/openworker` 2026-08-06 的 `permissions.py`、`risk.py`、`inbox.py`、`audit.py`、`selfwake.py`、`automation/models.py`、`mcp/tools.py` 为样本;项目仍在 beta,只借心智与不变量,不复制其本地单机 JSON/SQLite 实现。
**P0 决策候选——统一 `ActionPolicy`**:所有有副作用的工具在执行前经过一个机械决策点,风险至少分四类:`read`(读文件/知识库/外部查询,默认允许)、`workspace_mutation`(用户工作目录内写产物,默认允许,继续由沙箱/路径边界/产物机检兜底)、`external_effect`(发邮件/微信、MES/LIMS 写入、创建远端对象,默认 ask)、`destructive_or_privileged`(删除远端对象、扩大授权、修改持久许可,始终 ask)。**不照搬 OpenWorker 的“shell/本地写默认逐次问”**:zcbot 的 shell 在 per-user sandbox 内且文档生产会高频写工作区,逐次审批只会制造确认疲劳;本层按 blast radius 而非工具名字分类。风险元数据由平台可信注册表声明,不能信 skill、OpenAPI/MCP 描述或模型自报;未知外部工具 fail closed。可再提供 read-only discuss/plan 模式,但它必须由同一策略层机械拒绝副作用,不能只靠 prompt。
**P0 决策候选——统一 Attention Inbox**:`ask_user` 继续服务「2-4 个方向选择、结束本轮等下一条用户消息」的轻交互;新增 attention item 服务「某个在途动作暂停后从原 tool_call 恢复」。PG 是唯一事实源,最小状态机 `pending -> resolved|expired|cancelled`,以 `(task_id, tool_call_id)` 唯一保证重连/重启不重复提问,resolve 用条件更新实现 first-responder-wins。item 至少记录 kind(approval/question/notification)、脱敏请求摘要、resolution、来源渠道和时间Web SSE、企业微信/个人微信只是同一 item 的展示/响应 transport,不各存一份状态。删除 task、取消 run、授权被管理员撤回时确定性关闭关联 pending item恢复前重跑权限判断,防等待期间策略或用户权限已经变化。**边界**:不能把待审批工具参数作为普通 user 文本让模型重新解释,批准的是被冻结且可校验的具体动作;凭据和完整敏感正文不入 item。
**P0 决策候选——外部动作审计**:新增独立 `action_audit_events`(不挤进回答费用口径的 `usage_events`,也不拿 toolfail 代替),记录 user/task/run、provider/tool/operation、风险级别、决策依据、approval item/rule、执行状态、目标资源标识、脱敏 args/result preview 与时间。审计回答「谁在什么任务中、凭哪条授权、对哪个对象做了什么、结果如何」token/password/secret、邮件/消息正文、浏览器输入、外部响应全文机械脱敏或不落库。只读查询可按采样/高价值 operation 记,外部写与拒绝/审批必须全记。该项是 §8.14 从查询扩到写操作前的 hard prerequisite。
**P1——定时任务的精确目标长期授权**:无人值守任务不能靠“整个工具永远允许”。借鉴 OpenWorker 的 task-scoped standing rule,授权归具体 scheduled job,形态为 `provider + operation_id/tool + normalized_target`;删除/停用 job 或管理员撤权即失效。`normalized_target` 的组成字段由可信 provider definition 声明(如 recipient/channel_id/plant_id/dataset_id),模型不能自行挑字段,不支持通配符;创建任务时 consent card 同时展示将读取的数据与将写入的精确目标。shell、任意文件删除及无法提取稳定目标的动作不授长期许可,每次仍 ask。现有定时查询与确定性 notify 不受影响;只有未来开放外部写时才启用该契约。
**P1——显式 self-wake,修订 §8.13 的绝对边界但不改默认**:保留「后台进程完成后只通知人、不自动续跑」为默认;仅当用户明确要求连续科研闭环,或 agent 显式调用类似 `wake_on_process(proc_id)` 时,允许进程终态触发一次新 run,注入机械完成事件、exit code 和输出路径后继续分析。设每任务自动恢复次数、token/费用预算、截止时间与取消开关;日志仍由 agent 按需读,不把全文注入上下文;后续外部副作用照常进 Attention Inbox。目标场景是「启动模拟/拟合 -> 等完成 -> 检查收敛 -> 出图和结论」,不是通用 workflow/job graph。没有真实中间计算需求信号前不实施。
**P2——MCP 只作受控连接协议补充**:保留 §8.14 OpenAPI 元工具为院内 MES/ERP/LIMS 主入口MCP 用于确有需求的标准 SaaS/第三方服务。服务器地址和 OAuth/密钥由管理员 definition 管,普通用户不能填任意 URL启动/刷新时发现 schema 后仍过 pinned allowlist、风险元数据、响应体积与审计边界。**不照搬“一 MCP tool 一 JSON tool 全展开”**:继续复用 `external_system_search/call` 的延迟发现心智,把 MCP tool 映射为稳定 `provider/tool_id`,避免几十上百个 schema 常驻上下文和供应商新增能力后自动越权。协议内容、tool description 和返回值均是不可信数据。
**明确不借**:①Tauri 桌面壳、本地 secret store 和 JSON/SQLite 状态——OpenWorker 是个人单机,zcbot 是多用户 Web + PG + 蓝绿;②为展示广度铺 25+ 通用 SaaS connector——优先院内 MES/LIMS/设备/知识与企微的真实需求;③多 persona/多 agent 编排——职责隔离继续由 skill 承担,§6/§8.11 的证据门槛不变;④逐次审批沙箱 shell/工作区写入——确认疲劳且不增加外部 blast-radius 安全;⑤直接复制 beta 项目代码——并发、身份、持久化和恢复不变量不同。
**落地顺序/触发**:先 `ActionPolicy -> Attention Inbox(Web) -> 渠道响应 -> action audit`,四者作为一个完整外部写安全闭环;再按真实无人值守写需求增加 exact-target standing rule,按真实长计算中间步骤增加 self-wake有明确第三方 MCP 接入对象后再做 adapter。任何阶段都不得先开放外部写、再用 prompt 要求模型“记得询问”补安全边界。实现时需回写 §3.1 loop、§7.5 sandbox/tool registry、§8.5 scheduler、§8.7 channel、§8.13 bg proc 与 §8.14 external systems 的最终契约,并以 migration 保持现有 API/数据兼容。
--- ---
## 附录:DeepSeek V4 关键事实(2026-04-24) ## 附录:DeepSeek V4 关键事实(2026-04-24)

View File

@ -2,7 +2,7 @@
> 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9` > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`
最后更新:2026-08-05(通用 OpenAPI 外部系统与可配置认证,bump 0.62.0) 最后更新:2026-08-06(OpenAPI 请求体契约 + 大响应无损分段读取,bump 0.62.1)
--- ---
@ -21,6 +21,10 @@
## 已完成关键能力 ## 已完成关键能力
### 2026-08-06
- **08-06 / 0.62.1 / OpenAPI 请求体契约 + 大响应无损读取**:生产 task `506c6ad0` 暴露 Swagger 2 body `$ref` 未进入搜索结果,模型无法得知 BI dataset exec 需要 `body.query`,并在 400 只剩状态码时连续盲猜;现统一解析 Swagger 2/OpenAPI 3 请求体 schema返回限长脱敏的上游错误详情。外部响应不再按字节切成半截 JSON超出模型内联额度时完整写入 user/task 隔离的 24h 缓存并返回 `result_ref`,新增 JSON Pointer、分页和字段投影 reader单响应/task/user 分别以 10/50/200 MiB 守门;完整 479 项 unittest 全绿(17 skip),外部系统专项 26 项、Ruff、隔离 mypy、Python 编译与 diff 检查通过。无 schema、migration、依赖或 HTTP API 变化,未写生产 DB。
### 2026-08-05 ### 2026-08-05
- **08-05 / 0.62.0 / 通用 OpenAPI 外部系统 + 可配置认证**:将 Factory 专用运行态拆为通用 OpenAPI connector、认证 strategy 和 provider preset 注册表;标准 ERP/LIMS 等系统可直接在管理后台选择用户名密码换 Token、API Key 或 Bearer Token并配置 Token 字段与认证 Header无需再新增 Python connector。Factory MES 保持内置 preset、既有配置和旧 `{username,password}` HTTP 请求兼容;用户凭据表单按 definition 动态生成Swagger JSON 不落盘并按 `definition_id+user_id` 隔离缓存 5 分钟。完整 475 项 unittest 全绿(17 skip),相关 41 项回归、Python/JavaScript 语法及 diff 检查通过;当前环境无可用浏览器实例,真实页面点击留部署后冒烟。无 schema、migration 或依赖变化,未连接生产 DB。 - **08-05 / 0.62.0 / 通用 OpenAPI 外部系统 + 可配置认证**:将 Factory 专用运行态拆为通用 OpenAPI connector、认证 strategy 和 provider preset 注册表;标准 ERP/LIMS 等系统可直接在管理后台选择用户名密码换 Token、API Key 或 Bearer Token并配置 Token 字段与认证 Header无需再新增 Python connector。Factory MES 保持内置 preset、既有配置和旧 `{username,password}` HTTP 请求兼容;用户凭据表单按 definition 动态生成Swagger JSON 不落盘并按 `definition_id+user_id` 隔离缓存 5 分钟。完整 475 项 unittest 全绿(17 skip),相关 41 项回归、Python/JavaScript 语法及 diff 检查通过;当前环境无可用浏览器实例,真实页面点击留部署后冒烟。无 schema、migration 或依赖变化,未连接生产 DB。

View File

@ -1,3 +1,3 @@
# zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。 # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
# 改版本只动这一行。 # 改版本只动这一行。
__version__ = "0.62.0" __version__ = "0.62.1"

View File

@ -17,6 +17,7 @@ from urllib.parse import quote, urljoin, urlparse
import httpx import httpx
from .auth import ExternalAuthError, get_auth_strategy from .auth import ExternalAuthError, get_auth_strategy
from .results import MAX_STORED_RESULT_BYTES
class OpenApiError(RuntimeError): class OpenApiError(RuntimeError):
@ -26,6 +27,17 @@ class OpenApiError(RuntimeError):
_HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete") _HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete")
_SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} _SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
_SPEC_LOCK = Lock() _SPEC_LOCK = Lock()
_SCHEMA_MAX_DEPTH = 5
_SCHEMA_MAX_PROPERTIES = 50
_SCHEMA_MAX_ENUM_ITEMS = 30
_SCHEMA_MAX_NODES = 100
_SCHEMA_TEXT_MAX_CHARS = 1000
_ERROR_DETAIL_MAX_CHARS = 2000
_SENSITIVE_KEY_RE = re.compile(
r"(?:password|passwd|secret|token|api[_-]?key|authorization|cookie|credential)",
re.IGNORECASE,
)
def _bool_value(value: Any, default: bool) -> bool: def _bool_value(value: Any, default: bool) -> bool:
raw = str(value if value is not None else "").strip().lower() raw = str(value if value is not None else "").strip().lower()
@ -217,6 +229,189 @@ class OpenApiClient:
}) })
return results return results
@classmethod
def _compact_schema(
cls,
spec: dict[str, Any],
schema: Any,
*,
depth: int = 0,
seen_refs: frozenset[str] = frozenset(),
budget: Optional[list[int]] = None,
) -> dict[str, Any]:
"""展开本地 schema 引用并限制体量,供模型构造请求而非做完整规范校验。"""
if budget is None:
budget = [_SCHEMA_MAX_NODES]
if budget[0] <= 0:
return {"truncated": True}
budget[0] -= 1
if not isinstance(schema, dict) or depth > _SCHEMA_MAX_DEPTH:
return {}
ref = schema.get("$ref")
if isinstance(ref, str):
if ref in seen_refs:
return {"$ref": ref, "recursive": True}
prefixes = (
("#/definitions/", spec.get("definitions")),
(
"#/components/schemas/",
(spec.get("components") or {}).get("schemas"),
),
)
for prefix, definitions in prefixes:
if ref.startswith(prefix) and isinstance(definitions, dict):
target = definitions.get(ref[len(prefix) :])
if isinstance(target, dict):
return cls._compact_schema(
spec,
target,
depth=depth + 1,
seen_refs=seen_refs | {ref},
budget=budget,
)
return {"$ref": ref}
compact: dict[str, Any] = {}
for key in (
"type",
"title",
"description",
"format",
"default",
"nullable",
"x-nullable",
"minimum",
"maximum",
"minLength",
"maxLength",
"pattern",
):
value = schema.get(key)
if isinstance(value, str):
compact[key] = value[:_SCHEMA_TEXT_MAX_CHARS]
elif (
isinstance(value, (int, float, bool)) or value is None and key in schema
):
compact[key] = value
required = schema.get("required")
if isinstance(required, list):
compact["required"] = [
str(item) for item in required[:_SCHEMA_MAX_PROPERTIES]
]
enum = schema.get("enum")
if isinstance(enum, list):
compact["enum"] = [
item[:_SCHEMA_TEXT_MAX_CHARS] if isinstance(item, str) else item
for item in enum[:_SCHEMA_MAX_ENUM_ITEMS]
]
properties = schema.get("properties")
if isinstance(properties, dict) and depth < _SCHEMA_MAX_DEPTH:
compact["properties"] = {
str(name): cls._compact_schema(
spec,
value,
depth=depth + 1,
seen_refs=seen_refs,
budget=budget,
)
for name, value in list(properties.items())[:_SCHEMA_MAX_PROPERTIES]
if isinstance(value, dict)
}
items = schema.get("items")
if isinstance(items, dict) and depth < _SCHEMA_MAX_DEPTH:
compact["items"] = cls._compact_schema(
spec,
items,
depth=depth + 1,
seen_refs=seen_refs,
budget=budget,
)
additional = schema.get("additionalProperties")
if isinstance(additional, bool):
compact["additionalProperties"] = additional
elif isinstance(additional, dict) and depth < _SCHEMA_MAX_DEPTH:
compact["additionalProperties"] = cls._compact_schema(
spec,
additional,
depth=depth + 1,
seen_refs=seen_refs,
budget=budget,
)
for key in ("allOf", "anyOf", "oneOf"):
variants = schema.get(key)
if isinstance(variants, list) and depth < _SCHEMA_MAX_DEPTH:
compact[key] = [
cls._compact_schema(
spec,
item,
depth=depth + 1,
seen_refs=seen_refs,
budget=budget,
)
for item in variants[:10]
if isinstance(item, dict)
]
return compact
@classmethod
def _body_contract(
cls, spec: dict[str, Any], operation: dict[str, Any]
) -> Optional[dict[str, Any]]:
"""统一 Swagger 2 body parameter 与 OpenAPI 3 requestBody。"""
for param in operation.get("parameters") or []:
if isinstance(param, dict) and param.get("in") == "body":
return {
"parameter_name": param.get("name") or "body",
"required": bool(param.get("required")),
"content_type": "application/json",
"schema": cls._compact_schema(spec, param.get("schema")),
}
request_body = operation.get("request_body") or operation.get("requestBody")
if not isinstance(request_body, dict):
return None
content = request_body.get("content")
if not isinstance(content, dict) or not content:
return None
content_type = (
"application/json" if "application/json" in content else next(iter(content))
)
media = content.get(content_type)
if not isinstance(media, dict):
return None
return {
"required": bool(request_body.get("required")),
"content_type": content_type,
"schema": cls._compact_schema(spec, media.get("schema")),
}
@staticmethod
def _error_detail(response: httpx.Response) -> str:
"""返回限长、递归脱敏的上游错误详情,帮助模型停止盲猜参数。"""
def redact(value: Any) -> Any:
if isinstance(value, dict):
return {
str(key): "[REDACTED]"
if _SENSITIVE_KEY_RE.search(str(key))
else redact(item)
for key, item in value.items()
}
if isinstance(value, list):
return [redact(item) for item in value]
return value
try:
detail = json.dumps(
redact(response.json()), ensure_ascii=False, default=str
)
except ValueError:
detail = response.text.strip()
if not detail:
return ""
if len(detail) > _ERROR_DETAIL_MAX_CHARS:
detail = detail[:_ERROR_DETAIL_MAX_CHARS] + "...[truncated]"
return detail
def _spec_base_path(self, spec: dict[str, Any]) -> str: def _spec_base_path(self, spec: dict[str, Any]) -> str:
"""Return the API path prefix declared by Swagger 2 / OpenAPI 3. """Return the API path prefix declared by Swagger 2 / OpenAPI 3.
@ -325,6 +520,9 @@ class OpenApiClient:
} }
for p in op["parameters"] if isinstance(p, dict) and "$ref" not in p for p in op["parameters"] if isinstance(p, dict) and "$ref" not in p
] ]
body_contract = self._body_contract(spec, op)
if body_contract is not None:
compact["body"] = body_contract
compact.pop("request_body", None) compact.pop("request_body", None)
compact["recommended"] = recommended compact["recommended"] = recommended
scored.append(( scored.append((
@ -411,20 +609,25 @@ class OpenApiClient:
except httpx.HTTPError as exc: except httpx.HTTPError as exc:
raise OpenApiError(f"外部系统接口调用失败: {type(exc).__name__}") from exc raise OpenApiError(f"外部系统接口调用失败: {type(exc).__name__}") from exc
if response.status_code >= 400: if response.status_code >= 400:
raise OpenApiError(f"外部系统接口返回 HTTP {response.status_code}") detail = self._error_detail(response)
suffix = f": {detail}" if detail else ""
raise OpenApiError(f"外部系统接口返回 HTTP {response.status_code}{suffix}")
content_type = response.headers.get("content-type", "") content_type = response.headers.get("content-type", "")
try: try:
payload: Any = response.json() if "json" in content_type else response.text payload: Any = response.json() if "json" in content_type else response.text
except ValueError: except ValueError:
payload = response.text payload = response.text
encoded = json.dumps(payload, ensure_ascii=False, default=str) encoded = json.dumps(payload, ensure_ascii=False, default=str)
truncated = len(encoded.encode("utf-8")) > self.cfg.max_result_bytes response_bytes = len(encoded.encode("utf-8"))
if truncated: if response_bytes > MAX_STORED_RESULT_BYTES:
encoded = encoded.encode("utf-8")[: self.cfg.max_result_bytes].decode("utf-8", "ignore") raise OpenApiError(
payload = encoded f"外部系统响应超过安全下载上限({MAX_STORED_RESULT_BYTES} bytes)"
"请缩小查询范围或使用远端分页/聚合接口"
)
return { return {
"operation_id": operation_id, "operation_id": operation_id,
"status_code": response.status_code, "status_code": response.status_code,
"truncated": truncated, "truncated": False,
"response_bytes": response_bytes,
"data": payload, "data": payload,
} }

View File

@ -0,0 +1,258 @@
"""外部系统大响应的 task 私有缓存与结构化分段读取。"""
from __future__ import annotations
import json
import re
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from uuid import uuid4
from core.file_store import atomic_write_text
RESULT_CACHE_SUBDIR = ".zcbot_external_results"
RESULT_TTL_SECONDS = 24 * 60 * 60
MAX_STORED_RESULT_BYTES = 10 * 1024 * 1024
MAX_TASK_CACHE_BYTES = 50 * 1024 * 1024
MAX_USER_CACHE_BYTES = 200 * 1024 * 1024
_REF_RE = re.compile(r"^extres_[0-9a-f]{32}$")
_PREVIEW_MAX_DEPTH = 6
_PREVIEW_MAX_NODES = 120
_PREVIEW_MAX_DICT_KEYS = 30
_PREVIEW_MAX_LIST_ITEMS = 5
_PREVIEW_MAX_STRING_CHARS = 500
class ExternalResultError(RuntimeError):
pass
@dataclass(frozen=True)
class StoredExternalResult:
result_ref: str
original_bytes: int
expires_in_seconds: int
def _json_bytes(value: Any) -> bytes:
return json.dumps(value, ensure_ascii=False, default=str).encode("utf-8")
def _pointer_escape(value: str) -> str:
return value.replace("~", "~0").replace("/", "~1")
def build_result_preview(value: Any) -> tuple[Any, list[dict[str, Any]]]:
"""生成合法、定界的 JSON 预览,并列出可继续分段读取的位置。"""
reads: list[dict[str, Any]] = []
budget = [_PREVIEW_MAX_NODES]
def visit(item: Any, pointer: str, depth: int) -> Any:
if budget[0] <= 0:
reads.append({"json_pointer": pointer, "reason": "node_budget"})
return {"$preview_truncated": True}
budget[0] -= 1
if depth > _PREVIEW_MAX_DEPTH:
reads.append({"json_pointer": pointer, "reason": "depth"})
return {"$preview_truncated": True}
if isinstance(item, dict):
pairs = list(item.items())
shown = pairs[:_PREVIEW_MAX_DICT_KEYS]
result: Any = {
str(key): visit(
child,
pointer + "/" + _pointer_escape(str(key)),
depth + 1,
)
for key, child in shown
}
if len(pairs) > len(shown):
result["$preview_omitted_keys"] = len(pairs) - len(shown)
reads.append({
"json_pointer": pointer,
"reason": "keys",
"total_keys": len(pairs),
})
return result
if isinstance(item, list):
shown = item[:_PREVIEW_MAX_LIST_ITEMS]
result = [
visit(child, pointer + "/" + str(index), depth + 1)
for index, child in enumerate(shown)
]
if len(item) > len(shown):
reads.append({
"json_pointer": pointer,
"reason": "items",
"total_items": len(item),
"next_offset": len(shown),
})
return result
if isinstance(item, str) and len(item) > _PREVIEW_MAX_STRING_CHARS:
reads.append({
"json_pointer": pointer,
"reason": "string",
"total_chars": len(item),
})
return item[:_PREVIEW_MAX_STRING_CHARS] + "...[preview]"
return item
return visit(value, "", 0), reads
def resolve_json_pointer(value: Any, pointer: str) -> Any:
"""解析 RFC 6901 JSON Pointer空串表示完整根值。"""
pointer = (pointer or "").strip()
if not pointer:
return value
if not pointer.startswith("/"):
raise ExternalResultError("json_pointer 必须为空或以 / 开头")
current = value
for raw_part in pointer[1:].split("/"):
part = raw_part.replace("~1", "/").replace("~0", "~")
if isinstance(current, dict):
if part not in current:
raise ExternalResultError(f"json_pointer 不存在: {pointer}")
current = current[part]
elif isinstance(current, list):
try:
if not part.isdigit():
raise ValueError(part)
index = int(part)
current = current[index]
except (ValueError, IndexError) as exc:
raise ExternalResultError(f"json_pointer 不存在: {pointer}") from exc
else:
raise ExternalResultError(f"json_pointer 不存在: {pointer}")
return current
class ExternalResultStore:
def __init__(self, user_root: Path, task_id: str):
self.cache_root = Path(user_root) / RESULT_CACHE_SUBDIR
self.root = self.cache_root / str(task_id)
def _path(self, result_ref: str) -> Path:
if not _REF_RE.fullmatch(result_ref or ""):
raise ExternalResultError("result_ref 无效")
return self.root / f"{result_ref}.json"
def _sweep(self) -> None:
if not self.cache_root.is_dir():
return
cutoff = time.time() - RESULT_TTL_SECONDS
for path in self.cache_root.glob("*/extres_*.json"):
try:
if path.stat().st_mtime < cutoff:
path.unlink()
try:
path.parent.rmdir()
except OSError:
pass
except OSError:
continue
def store(self, system_id: str, result: dict[str, Any]) -> StoredExternalResult:
encoded = _json_bytes(result)
if len(encoded) > MAX_STORED_RESULT_BYTES:
raise ExternalResultError(
f"外部系统响应超过安全保存上限({MAX_STORED_RESULT_BYTES} bytes)"
)
self.root.mkdir(parents=True, exist_ok=True)
self._sweep()
existing = sorted(
self.root.glob("extres_*.json"),
key=lambda path: path.stat().st_mtime,
)
total = sum(path.stat().st_size for path in existing)
while existing and total + len(encoded) > MAX_TASK_CACHE_BYTES:
oldest = existing.pop(0)
try:
size = oldest.stat().st_size
oldest.unlink()
total -= size
except OSError:
continue
if total + len(encoded) > MAX_TASK_CACHE_BYTES:
raise ExternalResultError("当前任务的外部结果缓存已达上限")
user_files = sorted(
self.cache_root.glob("*/extres_*.json"),
key=lambda path: path.stat().st_mtime,
)
user_total = sum(path.stat().st_size for path in user_files)
while user_files and user_total + len(encoded) > MAX_USER_CACHE_BYTES:
oldest = user_files.pop(0)
try:
size = oldest.stat().st_size
oldest.unlink()
user_total -= size
except OSError:
continue
if user_total + len(encoded) > MAX_USER_CACHE_BYTES:
raise ExternalResultError("当前用户的外部结果缓存已达上限")
result_ref = f"extres_{uuid4().hex}"
envelope = {
"version": 1,
"created_at": int(time.time()),
"system_id": str(system_id),
"result": result,
}
atomic_write_text(self._path(result_ref), json.dumps(
envelope, ensure_ascii=False, default=str
))
return StoredExternalResult(
result_ref=result_ref,
original_bytes=len(encoded),
expires_in_seconds=RESULT_TTL_SECONDS,
)
def load(self, result_ref: str) -> dict[str, Any]:
self._sweep()
path = self._path(result_ref)
try:
if time.time() - path.stat().st_mtime > RESULT_TTL_SECONDS:
path.unlink()
raise ExternalResultError("result_ref 已过期")
payload = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise ExternalResultError("result_ref 不存在或已过期") from exc
except (OSError, ValueError) as exc:
raise ExternalResultError("result_ref 无法读取") from exc
if not isinstance(payload, dict) or not isinstance(payload.get("result"), dict):
raise ExternalResultError("result_ref 内容无效")
return payload
def project_result(
value: Any,
*,
offset: int = 0,
limit: int = 50,
fields: list[str] | None = None,
) -> dict[str, Any]:
"""对缓存节点分页并可选投影字典字段。"""
offset = max(0, int(offset))
limit = max(1, min(int(limit), 200))
if fields is not None and not isinstance(fields, list):
raise ExternalResultError("fields 必须是字符串数组")
wanted = [str(field) for field in (fields or [])[:50]]
def select(item: Any) -> Any:
if not wanted or not isinstance(item, dict):
return item
return {field: item.get(field) for field in wanted if field in item}
if isinstance(value, list):
page = [select(item) for item in value[offset : offset + limit]]
return {
"type": "array",
"total_items": len(value),
"offset": offset,
"limit": limit,
"has_more": offset + len(page) < len(value),
"data": page,
}
if isinstance(value, dict):
return {"type": "object", "data": select(value)}
return {"type": type(value).__name__, "data": value}

View File

@ -28,6 +28,7 @@ from tools.documents import DocumentDownloadTool, DocumentListKbTool, DocumentSe
from tools.external_systems import ( from tools.external_systems import (
ExternalSystemCallTool, ExternalSystemCallTool,
ExternalSystemListTool, ExternalSystemListTool,
ExternalSystemResultReadTool,
ExternalSystemSearchTool, ExternalSystemSearchTool,
) )
from tools.fs import EditTool, GlobTool, GrepTool, ReadTool, WriteTool from tools.fs import EditTool, GlobTool, GrepTool, ReadTool, WriteTool
@ -154,10 +155,22 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
] ]
def _external_systems() -> list: def _external_systems() -> list:
result_budget: dict[str, int] = {}
return [ return [
ExternalSystemListTool(ctx.uid, **base), ExternalSystemListTool(ctx.uid, **base),
ExternalSystemSearchTool(ctx.uid, **base), ExternalSystemSearchTool(ctx.uid, **base),
ExternalSystemCallTool(ctx.uid, **base), ExternalSystemCallTool(
ctx.uid,
task_id=ctx.task_id,
result_budget=result_budget,
**wd_base,
),
ExternalSystemResultReadTool(
ctx.uid,
task_id=ctx.task_id,
result_budget=result_budget,
**wd_base,
),
] ]
def _load_skill() -> list: def _load_skill() -> list:

View File

@ -3,6 +3,7 @@ from __future__ import annotations
import json import json
import os import os
import sys import sys
import tempfile
import unittest import unittest
import uuid import uuid
from copy import deepcopy from copy import deepcopy
@ -59,6 +60,16 @@ def _cfg(*, allowed=frozenset(), recommended=()):
_SPEC = { _SPEC = {
"swagger": "2.0", "swagger": "2.0",
"definitions": {
"DataExec": {
"type": "object",
"required": ["query"],
"properties": {
"query": {"title": "查询字典参数", "type": "object"},
"is_test": {"type": "boolean", "default": False},
},
}
},
"paths": { "paths": {
"/api/bi/dataset/": { "/api/bi/dataset/": {
"get": { "get": {
@ -85,7 +96,12 @@ _SPEC = {
"summary": "执行只读数据集", "summary": "执行只读数据集",
"parameters": [ "parameters": [
{"name": "code", "in": "path", "required": True, "type": "string"}, {"name": "code", "in": "path", "required": True, "type": "string"},
{"name": "payload", "in": "body", "required": True, "schema": {"type": "object"}}, {
"name": "payload",
"in": "body",
"required": True,
"schema": {"$ref": "#/definitions/DataExec"},
},
], ],
} }
}, },
@ -236,6 +252,70 @@ class FactoryOpenApiConnectorTests(unittest.TestCase):
) )
self.assertTrue(all(item["recommended"] for item in result[:2])) self.assertTrue(all(item["recommended"] for item in result[:2]))
def test_search_includes_resolved_swagger_body_schema(self):
from core.external_systems.factory import FactoryMesClient
http = _Http()
client = FactoryMesClient(
"mes-user", "mes-password", _cfg(allowed={"bi_dataset_exec"})
)
with patch.object(client, "_client", return_value=http):
result = client.search("执行只读数据集")
operation = next(
item for item in result if item["operation_id"] == "bi_dataset_exec"
)
self.assertEqual(operation["body"]["parameter_name"], "payload")
self.assertTrue(operation["body"]["required"])
self.assertEqual(
operation["body"]["schema"]["properties"]["query"]["type"],
"object",
)
self.assertEqual(operation["body"]["schema"]["required"], ["query"])
def test_search_includes_openapi_3_request_body_schema(self):
from core.external_systems.openapi import OpenApiClient
spec = {
"openapi": "3.0.0",
"components": {
"schemas": {
"ReportQuery": {
"type": "object",
"properties": {"month": {"type": "string"}},
}
}
},
"paths": {
"/reports/preview/": {
"post": {
"operationId": "report_preview",
"summary": "预览报表",
"requestBody": {
"required": True,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ReportQuery"
}
}
},
},
}
}
},
}
client = OpenApiClient({"token": "private"}, _cfg(allowed={"report_preview"}))
with (
patch.object(client, "authenticate", return_value={}),
patch.object(client, "_fetch_spec", return_value=spec),
):
result = client.search("预览报表")
self.assertEqual(result[0]["body"]["content_type"], "application/json")
self.assertEqual(
result[0]["body"]["schema"]["properties"]["month"]["type"],
"string",
)
def test_get_call_resolves_encoded_path_and_query(self): def test_get_call_resolves_encoded_path_and_query(self):
from core.external_systems.factory import FactoryMesClient from core.external_systems.factory import FactoryMesClient
@ -389,12 +469,60 @@ class FactoryOpenApiConnectorTests(unittest.TestCase):
result = allowed.call( result = allowed.call(
"bi_dataset_exec", "bi_dataset_exec",
arguments={"code": "quality"}, arguments={"code": "quality"},
body={"batch": "B-1"}, body={"query": {"batch": "B-1"}},
) )
request = [call for call in http.calls if call[0] == "POST" and "/dataset/" in call[1]][0] request = [call for call in http.calls if call[0] == "POST" and "/dataset/" in call[1]][0]
self.assertEqual(request[2]["json"], {"batch": "B-1"}) self.assertEqual(request[2]["json"], {"query": {"batch": "B-1"}})
self.assertFalse(result["truncated"]) self.assertFalse(result["truncated"])
def test_call_preserves_payload_larger_than_inline_limit(self):
from core.external_systems.factory import FactoryMesClient
class LargeHttp(_Http):
def request(self, method, url, **kwargs):
self.calls.append((method, url, kwargs))
return _Response(payload={"rows": "x" * 70000})
client = FactoryMesClient(
"u", "p", _cfg(allowed={"bi_dataset_exec"})
)
with patch.object(client, "_client", return_value=LargeHttp()):
result = client.call(
"bi_dataset_exec",
arguments={"code": "quality"},
body={"query": {}},
)
self.assertEqual(len(result["data"]["rows"]), 70000)
self.assertGreater(result["response_bytes"], client.cfg.max_result_bytes)
self.assertFalse(result["truncated"])
def test_call_surfaces_sanitized_upstream_error_detail(self):
from core.external_systems.factory import FactoryMesClient, FactoryMesError
class ErrorHttp(_Http):
def request(self, method, url, **kwargs):
self.calls.append((method, url, kwargs))
return _Response(
status_code=400,
payload={
"query": ["This field is required."],
"access_token": "must-not-leak",
},
)
client = FactoryMesClient("u", "p", _cfg(allowed={"bi_dataset_exec"}))
with patch.object(client, "_client", return_value=ErrorHttp()):
with self.assertRaises(FactoryMesError) as raised:
client.call(
"bi_dataset_exec",
arguments={"code": "yield"},
body={"wrong": "shape"},
)
message = str(raised.exception)
self.assertIn("This field is required.", message)
self.assertIn("[REDACTED]", message)
self.assertNotIn("must-not-leak", message)
def test_rejects_unknown_arguments(self): def test_rejects_unknown_arguments(self):
from core.external_systems.factory import FactoryMesClient, FactoryMesError from core.external_systems.factory import FactoryMesClient, FactoryMesError
@ -446,23 +574,64 @@ class ExternalSystemToolSafetyTests(unittest.TestCase):
self.assertEqual(payload["query_guidance"], "统计查询先查看数据集目录") self.assertEqual(payload["query_guidance"], "统计查询先查看数据集目录")
self.assertEqual(payload["recommended_operation_ids"], ["bi_dataset_list"]) self.assertEqual(payload["recommended_operation_ids"], ["bi_dataset_list"])
def test_call_discards_result_over_per_run_external_budget(self): def test_large_call_spills_and_result_reader_pages_without_data_loss(self):
from tools.external_systems import ExternalSystemCallTool from tools.external_systems import (
ExternalSystemCallTool,
ExternalSystemResultReadTool,
)
uid = uuid.uuid4() uid = uuid.uuid4()
task_id = uuid.uuid4()
system_id = str(uuid.uuid4())
budget: dict[str, int] = {}
client = SimpleNamespace( client = SimpleNamespace(
cfg=SimpleNamespace(max_total_result_bytes=32), cfg=SimpleNamespace(max_result_bytes=512, max_total_result_bytes=4096),
call=lambda *args, **kwargs: {"data": "x" * 100}, call=lambda *args, **kwargs: {
"operation_id": "detail_list",
"status_code": 200,
"truncated": False,
"data": [{"id": index, "noise": "x" * 100} for index in range(20)],
},
) )
with patch( with tempfile.TemporaryDirectory() as tmp, patch(
"tools.external_systems._row_and_client", "tools.external_systems._row_and_client",
return_value=(SimpleNamespace(), client), return_value=(SimpleNamespace(), client),
): ):
output = ExternalSystemCallTool(uid).execute( call_tool = ExternalSystemCallTool(
str(uuid.uuid4()), "detail_list" uid,
task_id=task_id,
result_budget=budget,
base_dir=Path(tmp),
) )
self.assertIn("累计返回量超过上限", output) spilled = json.loads(call_tool.execute(system_id, "detail_list"))
self.assertNotIn("x" * 20, output) self.assertFalse(spilled["inline_complete"])
self.assertFalse(spilled["truncated"])
self.assertRegex(spilled["result_ref"], r"^extres_[0-9a-f]{32}$")
self.assertNotIn("x" * 100, json.dumps(spilled))
read_tool = ExternalSystemResultReadTool(
uid,
task_id=task_id,
result_budget=budget,
base_dir=Path(tmp),
)
page = json.loads(read_tool.execute(
spilled["result_ref"],
json_pointer="/data",
offset=5,
limit=2,
fields=["id"],
))
other_task = ExternalSystemResultReadTool(
uid,
task_id=uuid.uuid4(),
base_dir=Path(tmp),
)
cross_task = other_task.execute(spilled["result_ref"])
self.assertEqual(page["total_items"], 20)
self.assertEqual(page["data"], [{"id": 5}, {"id": 6}])
self.assertTrue(page["has_more"])
self.assertIn("不存在或已过期", cross_task)
if __name__ == "__main__": if __name__ == "__main__":

View File

@ -5,6 +5,13 @@ import json
from uuid import UUID from uuid import UUID
from core.external_systems.factory import FactoryMesError from core.external_systems.factory import FactoryMesError
from core.external_systems.results import (
ExternalResultError,
ExternalResultStore,
build_result_preview,
project_result,
resolve_json_pointer,
)
from core.external_systems.service import ( from core.external_systems.service import (
ExternalSystemError, ExternalSystemError,
client_for_external_system, client_for_external_system,
@ -86,6 +93,7 @@ class ExternalSystemCallTool(Tool):
description = ( description = (
"调用已连接外部系统的受控只读 OpenAPI operation。必须使用 search 返回的 operation_id" "调用已连接外部系统的受控只读 OpenAPI operation。必须使用 search 返回的 operation_id"
"不接受 URL。GET/HEAD 默认可用POST 仅限管理员声明的只读 operation。" "不接受 URL。GET/HEAD 默认可用POST 仅限管理员声明的只读 operation。"
"大响应会完整保存并返回 result_ref使用 external_system_result_read 分段读取。"
) )
parameters = { parameters = {
"type": "object", "type": "object",
@ -99,16 +107,72 @@ class ExternalSystemCallTool(Tool):
}, },
"body": { "body": {
"type": "object", "type": "object",
"description": "仅对管理员放行的只读 POST 操作提供 JSON body" "description": (
"仅对管理员放行的只读 POST 操作提供原始 JSON 请求体;"
"严格遵循 search 返回的 body.schema不要按 Swagger body 参数名再包一层"
)
}, },
}, },
"required": ["system_id", "operation_id"], "required": ["system_id", "operation_id"],
} }
def __init__(self, user_id: UUID, **kwargs): def __init__(
self,
user_id: UUID,
*,
task_id: UUID | str = "default",
result_budget: dict[str, int] | None = None,
**kwargs,
):
super().__init__(**kwargs) super().__init__(**kwargs)
self.user_id = user_id self.user_id = user_id
self._result_bytes: dict[str, int] = {} self._result_bytes = result_budget if result_budget is not None else {}
self._result_store = ExternalResultStore(
self.user_root or self.base_dir, str(task_id)
)
def _bounded_output(
self,
system_id: str,
result: dict,
*,
per_result_limit: int,
total_limit: int,
) -> str:
rendered = _json(result)
used = self._result_bytes.get(system_id, 0)
remaining = max(0, total_limit - used)
if len(rendered.encode("utf-8")) <= min(per_result_limit, remaining):
self._result_bytes[system_id] = used + len(rendered.encode("utf-8"))
return rendered
stored = self._result_store.store(system_id, result)
preview, reads = build_result_preview(result)
envelope = {
"operation_id": result.get("operation_id"),
"status_code": result.get("status_code"),
"truncated": False,
"inline_complete": False,
"result_ref": stored.result_ref,
"original_bytes": stored.original_bytes,
"expires_in_seconds": stored.expires_in_seconds,
"preview": preview,
"available_reads": reads,
}
output = _json(envelope)
inline_limit = min(per_result_limit, remaining)
if len(output.encode("utf-8")) > inline_limit:
envelope["preview"] = None
envelope["available_reads"] = reads[:20]
output = _json(envelope)
if len(output.encode("utf-8")) > remaining:
self._result_bytes[system_id] = total_limit
return (
"[Error] 本轮外部系统内联返回量已达上限。完整结果已缓存为 "
f"{stored.result_ref},请在下一轮使用 external_system_result_read 读取。"
)
self._result_bytes[system_id] = used + len(output.encode("utf-8"))
return output
def execute( def execute(
self, self,
@ -123,19 +187,116 @@ class ExternalSystemCallTool(Tool):
used = self._result_bytes.get(system_id, 0) used = self._result_bytes.get(system_id, 0)
if used >= client.cfg.max_total_result_bytes: if used >= client.cfg.max_total_result_bytes:
return ( return (
"[Error] 本轮外部系统返回量已达上限。请改用聚合接口或 dataset" "[Error] 本轮外部系统内联返回量已达上限。"
"不要继续分页拉取日志/明细" "请在下一轮继续查询,或读取之前返回的 result_ref"
) )
result = client.call(operation_id, arguments=arguments, body=body) result = client.call(operation_id, arguments=arguments, body=body)
result_size = len(_json(result).encode("utf-8")) return self._bounded_output(
used += result_size system_id,
self._result_bytes[system_id] = used result,
if used > client.cfg.max_total_result_bytes: per_result_limit=client.cfg.max_result_bytes,
return ( total_limit=client.cfg.max_total_result_bytes,
"[Error] 本轮外部系统累计返回量超过上限,当前结果已丢弃。" )
"请改用聚合接口或 dataset并缩小查询范围。" except (ExternalSystemError, FactoryMesError, ExternalResultError) as exc:
)
return _json(result)
except (ExternalSystemError, FactoryMesError) as exc:
print(f"[WARN] external system call failed: {type(exc).__name__}") print(f"[WARN] external system call failed: {type(exc).__name__}")
return f"[Error] {exc}" return f"[Error] {exc}"
class ExternalSystemResultReadTool(Tool):
name = "external_system_result_read"
description = (
"分段读取 external_system_call 返回的完整大结果。result_ref 仅对当前 task 有效;"
"使用 JSON Pointer 定位数组或对象,并用 offset/limit/fields 控制返回量。"
)
parameters = {
"type": "object",
"properties": {
"result_ref": {"type": "string"},
"json_pointer": {
"type": "string",
"default": "",
"description": "RFC 6901 JSON Pointer空字符串表示根节点",
},
"offset": {"type": "integer", "minimum": 0, "default": 0},
"limit": {"type": "integer", "minimum": 1, "maximum": 200, "default": 50},
"fields": {
"type": "array",
"items": {"type": "string"},
"maxItems": 50,
"description": "目标是对象或对象数组时,仅返回这些字段",
},
},
"required": ["result_ref"],
}
def __init__(
self,
user_id: UUID,
*,
task_id: UUID | str = "default",
result_budget: dict[str, int] | None = None,
**kwargs,
):
super().__init__(**kwargs)
self.user_id = user_id
self._result_bytes = result_budget if result_budget is not None else {}
self._result_store = ExternalResultStore(
self.user_root or self.base_dir, str(task_id)
)
def execute(
self,
result_ref: str,
json_pointer: str = "",
offset: int = 0,
limit: int = 50,
fields: list[str] | None = None,
**kwargs,
) -> str:
try:
stored = self._result_store.load(result_ref)
system_id = str(stored.get("system_id") or "")
_, client = _row_and_client(self.user_id, system_id)
used = self._result_bytes.get(system_id, 0)
remaining = client.cfg.max_total_result_bytes - used
if remaining <= 0:
return "[Error] 本轮外部系统内联返回量已达上限,请在下一轮继续读取。"
selected = resolve_json_pointer(stored["result"], json_pointer)
response = {
"result_ref": result_ref,
"json_pointer": json_pointer,
**project_result(selected, offset=offset, limit=limit, fields=fields),
}
output = _json(response)
inline_limit = min(client.cfg.max_result_bytes, remaining)
if len(output.encode("utf-8")) > inline_limit:
preview, reads = build_result_preview(response)
output = _json({
"result_ref": result_ref,
"json_pointer": json_pointer,
"inline_complete": False,
"preview": preview,
"available_reads": reads,
"hint": "减小 limit、指定更深的 json_pointer 或使用 fields 投影",
})
if len(output.encode("utf-8")) > inline_limit:
output = _json({
"result_ref": result_ref,
"json_pointer": json_pointer,
"inline_complete": False,
"hint": "当前分段仍过大;请减小 limit、指定更深的 json_pointer 或使用 fields",
})
if len(output.encode("utf-8")) > remaining:
self._result_bytes[system_id] = client.cfg.max_total_result_bytes
return "[Error] 本轮外部系统内联返回量已达上限,请在下一轮继续读取。"
self._result_bytes[system_id] = used + len(output.encode("utf-8"))
return output
except (
ExternalSystemError,
FactoryMesError,
ExternalResultError,
TypeError,
ValueError,
) as exc:
print(f"[WARN] external system result read failed: {type(exc).__name__}")
return f"[Error] {exc}"