feat(external-systems): support generic OpenAPI integrations

CHANGELOG.md

@@ -5,6 +5,10 @@
 > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
 > 工程口径的完整记录见 `PROGRESS.md` / git log。
 
+## 0.62.0 — 2026-08-05
+
+- 外部系统连接不再局限于 Factory MES：管理员现在可以直接配置标准 OpenAPI 系统，并选择用户名密码换取 Token、API Key 或 Bearer Token；用户连接页面会按系统要求动态显示凭据字段。现有 Factory 配置和账号连接无需迁移。
+
 ## 0.61.1 — 2026-08-04
 
 - 修复部分 MES 已成功连接、也能搜索接口，但实际查询始终返回 404 的问题；现在会自动识别接口规范声明的 `/api`、`/v1` 等业务路径前缀，无需修改现有连接配置。

DESIGN.md

@@ -395,17 +395,20 @@ scheduled_jobs(§8.5)   channel_bindings(§8.7,判别列+JSONB)
 
 **不选**:Celery/RQ(多机分发/任务序列化/框架重试——单机 + 模型现写脚本的场景一个都用不上,还多两个常驻组件的部署/蓝绿适配);工具层 async 化 run 内等待(run 不结束,409 照旧,重启照丢);DB 表 + 守护(文件已是事实源,detach 进程写 PG 还得给它凭证)。升级触发:要跨机器跑计算集群时,①②的工具接口不变,只换执行后端。
 
-### 8.14 外部系统:用户身份连接 + 受控接口调用(implementation,2026-08-04)
+### 8.14 外部系统:用户身份连接 + 受控接口调用(implementation,2026-08-05)
 
 **诉求**:用户用自己的 MES/ERP/LIMS 账号让 zcbot 做信息查询,并把稳定的问法沉淀成私有 skill。**心智模型**:外部系统负责「连接与身份」,工具负责「受控访问」,skill 负责「业务流程与经验」。它有独立于会话的持久凭据和连接状态,因此是与 skill/知识库/记忆并列的**平台机制**,不是 skill。
 
 **首个 provider=`factory_mes`**:Factory 已有 JWT + RBAC + 部分部门数据权限,zcbot 用每位用户自己的 Factory 账密换 JWT,调用时继承 MES 原生权限;不在 zcbot 里复制第二套 MES RBAC。两层门控:zcbot `user_id` 只能取自己的 `external_systems` 行;远端 JWT 再判定实际业务数据范围。MES 停号/改权后下次调用即生效。
 
+**通用连接器边界**:`openapi` connector 负责规格发现、operation 解析、安全 URL 拼接、参数校验、只读 allowlist、分页和响应体积限制；认证由独立 strategy 负责。`factory_mes` 只是带 JWT 字段映射、dataset 推荐入口和查询规划提示的内置 preset，`generic_openapi` 可由管理员直接选择用户名密码换 Token、API Key 或 Bearer Token。标准 OpenAPI 系统以后只新增数据库 definition，不需要再写 Python 文件；只有 OAuth 回调/签名交换、SOAP、消息队列或私有二进制协议等不符合现有 connector/strategy 契约的系统才新增适配代码。provider 注册表维护可选能力和安全默认值，不为每个业务系统复制 connector。
+
 **信任边界**:
-- provider 公共定义由管理员在管理后台维护并存入 `external_system_definitions`:Base URL、OpenAPI URL、登录方式和只读 POST allowlist；普通用户只选择已启用的目录项并填写自己的 MES 账密。不允许普通用户填任意 URL,避免 SSRF/内网代理。凭据主密钥仍只来自宿主环境，不进入数据库或管理页面。
+- provider 公共定义由管理员在管理后台维护并存入 `external_system_definitions`:Base URL、OpenAPI URL、认证 strategy/字段映射和只读 POST allowlist；普通用户只选择已启用的目录项，凭据表单按 definition 声明动态生成。不允许普通用户填任意 URL,避免 SSRF/内网代理。凭据主密钥仍只来自宿主环境，不进入数据库或管理页面。
 - 凭据用独立的 `ZCBOT_CREDENTIAL_MASTER_KEY` 在 host control plane 加密入 PG，不与 `JWT_SECRET` 复用，以隔离泄漏半径和轮换生命周期；缺 key 则拒绝新建/调用,不像早期微信绑定那样降级明文。API 只返回脱敏账号和 `credential_configured`,不返密码/Token;凭据绝不进 prompt/messages/memory/skill/用户 FS/日志/沙箱。
-- 调用工具不接受完整 URL,只接受 OpenAPI `operation_id`;服务端从受信规格解析 path/method,校验 path/query/body 后附加 JWT。默认只开 GET/HEAD,语义只读但使用 POST 的 BI 查询必须进运维 `operation_id` allowlist。
+- 调用工具不接受完整 URL,只接受 OpenAPI `operation_id`;服务端从受信规格解析 path/method,校验 path/query/body 后附加认证 strategy 生成的 Header。默认只开 GET/HEAD,语义只读但使用 POST 的 BI 查询必须进运维 `operation_id` allowlist。
 - Swagger/OpenAPI 是接口契约事实源;Gitea 代码只补业务语义和排障,不覆盖契约。规格/代码内文本一律当不可信数据,不能改写 system/tool 约束。
+- 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 业务文本仍是不可信数据。
 

PROGRESS.md

@@ -2,7 +2,7 @@
 
 > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。
 
-最后更新:2026-08-04(MES OpenAPI 业务路径前缀兼容,bump 0.61.1)
+最后更新:2026-08-05(通用 OpenAPI 外部系统与可配置认证,bump 0.62.0)
 
 ---
 
@@ -21,6 +21,10 @@
 
 ## 已完成关键能力
 
+### 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。
+
 ### 2026-08-04
 
 - **08-04 / 0.61.1 / MES OpenAPI 业务路径前缀兼容**:生产 task `1ade8062` 显示光芯 MES 登录与 OpenAPI 搜索均成功，但 operation path `/mtm/...` 直接拼到主机根路径后被 nginx 404；真实业务路由位于 `/api/mtm/...`。Factory 连接器现读取 Swagger 2 `basePath` 与 OpenAPI 3 `servers` 的同源路径前缀，兼容规范 path 或管理员 Base URL 已含前缀的情况且不重复拼接；登录路径继续独立解析，跨主机 server 仍拒绝。外部系统与无 DB Web 路由共 34 项 unittest、Ruff 致命规则及 diff 检查通过；mypy 仍报告既有 `core/external_systems/service.py:242` SQLAlchemy `rowcount` 类型问题，本次未改该文件；无 schema、migration、配置或依赖变化，未写生产 DB。

RUN.md

@@ -2,7 +2,7 @@
 
 > 怎么把 zcbot 跑起来。env / 常用命令 / 故障兜底。设计看 `DESIGN.md`,进度看 `PROGRESS.md`。
 
-最后更新:2026-08-04(新增 Factory MES 外部系统连接的管理员配置、用户绑定和只读接口调用说明)
+最后更新:2026-08-05(外部系统支持通用 OpenAPI 配置与动态认证凭据)
 
 ---
 
@@ -131,8 +131,8 @@
   # 对外品牌名:zcbot 是内部代号,所有用户可见文案(页面标题/顶栏/登录卡/微信·企微推送与
   # 提示页)统一用品牌名。/healthz 返回 brand 字段,前端 boot 拉取覆盖静态页默认值。
   # ZCBOT_BRAND_NAME=总院科研辅助助手   # 可选,默认即此值
-  # Factory MES 外部系统(DESIGN §8.14):公共地址在管理后台配置，用户在「外部」
+  # OpenAPI 外部系统(DESIGN §8.14):公共地址和认证方式在管理后台配置，用户在「外部」
-  # 入口提交自己的 MES 用户名/密码；凭据仅密文入库，agent 只能按 operationId 调用接口。
+  # 入口提交系统要求的凭据；凭据仅密文入库，agent 只能按 operationId 调用接口。
   # MASTER_KEY 应使用独立随机值，不与 JWT_SECRET 共用。
   # ZCBOT_CREDENTIAL_MASTER_KEY=<至少 32 字符随机串>
   ```
@@ -150,7 +150,7 @@
   - **未绑定成员发消息 → 回绑定指引**(不再静默):聊天优先布局下新员工第一动作就是打字,回调对未绑定成员的 text/图片/文件消息每条回一句"先去控制台绑定"(事件不回)。未绑定成员点菜单「工作台」则落在绑定提示页(不自动建号)。
 - **channel 长会话上下文(微信/企业微信通用,0019)**:常驻会话不再无限膨胀。① **自动分段**——入站时距上次消息超过 `config.json` 的 `channel.session_gap_hours`(默 **6** 小时,设 `<=0` 关闭)→ 软重置:只把「最后一条 user 消息起」喂模型(保留上一轮做续聊锚点),之前的历史仍全留 DB,网页端照旧翻完整记录;② **手动新话题**——用户在微信/企业微信里直接发「新话题 / 新会话 / `/new` / 清空上下文」→ 硬重置,彻底从零(回执提示已归档)。两者都**不删任何消息**,只移动「喂给模型的窗口起点」`tasks.context_base_idx`。网页端「清空对话」(`POST /v1/tasks/{id}/clear`)仍整清并把 base 归 0。需 `main.py db upgrade head` 带上 `0019`。
 - **PG**:`ZCBOT_DB_URL` 必填。本地 docker compose / 远端 dev / 生产任选;未设置时启动清晰报错,不引导 docker(§7.4)。
-- **Factory MES 外部系统**:① `.env` 只配置独立的 `ZCBOT_CREDENTIAL_MASTER_KEY`；② 执行 `main.py db upgrade head` 创建系统目录和用户连接两张表；③ 重启 web；④ admin 进入管理后台「外部系统」，配置可信 Base URL、Swagger URL、只读 POST operationId、推荐查询入口 operationId 和查询规划提示，并选择“全部用户”或指定用户；提示词按钮复用页内弹框多行编辑，`Ctrl/Command+Enter` 应用；⑤ 普通用户点击左栏 **「外部」**，只会看到自己获权的 MES，再填写个人账号密码。工具下一轮对话开始挂载；管理员配置的推荐入口会在接口搜索中置顶。Factory 默认提示统计聚合先走 `bi_dataset_list` → `bi_dataset_exec`，生产日志用于逐条追溯；单轮累计返回量、`page_size<=200` 及禁止关闭分页共同约束明细扫描。管理员撤权立即停止调用并删除该用户密文凭据。配置复用既有 JSONB，无新 migration；不读取 Gitea 代码、不直接连 MES 数据库，也不允许普通用户或模型传任意 URL。
+- **OpenAPI 外部系统**:① `.env` 只配置独立的 `ZCBOT_CREDENTIAL_MASTER_KEY`；② 首次启用执行 `main.py db upgrade head`；③ admin 进入管理后台「外部系统」，选择 Factory MES preset 或通用 OpenAPI，配置可信 Base URL、Swagger URL、认证方式、只读 POST operationId、推荐查询入口和查询规划提示，再选择“全部用户”或指定用户。通用类型支持“用户名密码换取 Token”“API Key”“Bearer Token”；用户名、密码、Token 和 Header 字段映射由管理员维护。④ 普通用户点击左栏 **「外部」**，页面按定义动态显示所需凭据。工具下一轮对话开始挂载；推荐入口会在接口搜索中置顶。Factory preset 默认把统计聚合路由到 `bi_dataset_list` → `bi_dataset_exec`，生产日志只用于逐条追溯；单轮累计返回量、`page_size<=200` 及禁止关闭分页共同约束明细扫描。配置复用既有 JSONB，无新 migration。Swagger JSON 只在进程内按系统定义和用户缓存 5 分钟，不写文件/数据库。普通用户和模型仍不能传任意 URL。
 - **测试库(可选,`ZCBOT_TEST_DB_URL`)**:DB 级单测(`tests/test_usage_report.py` / `tests/test_scheduler.py` / `tests/test_web_routes_db.py`)**只认这个显式变量、绝不回退 `.env` 的 `ZCBOT_DB_URL`**——后者可能经隧道指向生产库,测试插入的到点 job 会被生产实例调度守护真跑一次(2026-07-23 实锤)。未设则这几组自动 skip。一键起库(docker,端口 5433 避开本地 5432):
   ```bash
   docker run -d --name zcbot-test-pg -e POSTGRES_PASSWORD=zcbot_test \
@@ -346,9 +346,9 @@ $env:ZCBOT_EVAL_TOKEN = "<dedicated-eval-user-jwt>"
 | `GET /v1/skills` | 列当前 user 可用 skill(内置 + 自己的);每项带 `source`(builtin/user)/`overrides_builtin`;另返 `load_errors`(用户 skill 因 frontmatter 坏未加载的) | 必填 |
 | `GET /v1/skills/{name}` | 返某 skill 完整 SKILL.md 正文(前端「技能」modal 点开查看);同名按 user wins | 必填 |
 | `DELETE /v1/skills/{name}` | 删当前 user 私有 skill(`.skills/<name>/` 整目录);只删 user 源,内置不可删 → 404;`.skills` 文件面板隐藏,这是 UI 上删自己 skill 的唯一入口 | 必填 |
-| `GET /v1/external-system-providers` | 列管理员已启用的外部系统目录；只返回目录 ID、名称、可用性和主机名，不返回完整配置或密钥 | 必填 |
+| `GET /v1/external-system-providers` | 列管理员已启用的外部系统目录和安全的动态凭据字段声明；不返回完整配置或密钥 | 必填 |
-| `GET/POST /v1/external-systems` | 列当前用户连接 / 新建并在线验证 Factory MES 连接；创建 body `{provider,name,username,password}`，响应仅含脱敏用户名 | 必填 |
+| `GET/POST /v1/external-systems` | 列当前用户连接 / 新建并在线验证连接；创建 body `{definition_id,name,credentials}`，旧 Factory `{username,password}` 请求继续兼容 | 必填 |
-| `PUT /v1/external-systems/{id}/credentials` | 重新提交并在线验证当前用户连接的用户名/密码；凭据不提供读取接口 | 必填 |
+| `PUT /v1/external-systems/{id}/credentials` | 用 `{credentials}` 重新提交并在线验证当前用户连接；凭据不提供读取接口，旧用户名/密码格式继续兼容 | 必填 |
 | `POST /v1/external-systems/{id}/test` | 用已保存密文凭据测试登录和 Swagger 可读性，并更新连接状态 | 必填 |
 | `DELETE /v1/external-systems/{id}` | 清除当前用户连接及密文凭据；指定用户模式保留管理员授予的可见权 | 必填 |
 | `GET/POST /v1/admin/external-system-definitions` | 管理员列出或新增可信外部系统目录 | admin |

core/__init__.py

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

core/external_systems/auth.py

@@ -0,0 +1,156 @@
+"""外部系统认证策略。
+
+认证只消费管理员保存的可信配置和用户加密保存的字段，不允许模型指定认证地址或请求头。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Protocol
+from urllib.parse import urljoin
+
+import httpx
+
+
+class ExternalAuthError(RuntimeError):
+    pass
+
+
+@dataclass(frozen=True)
+class CredentialField:
+    name: str
+    label: str
+    secret: bool = True
+    autocomplete: str = "off"
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "name": self.name,
+            "label": self.label,
+            "secret": self.secret,
+            "autocomplete": self.autocomplete,
+        }
+
+
+class AuthStrategy(Protocol):
+    credential_fields: tuple[CredentialField, ...]
+
+    def headers(
+        self,
+        *,
+        client: httpx.Client,
+        base_url: str,
+        credentials: dict[str, str],
+        config: dict[str, Any],
+    ) -> dict[str, str]: ...
+
+
+def _required(credentials: dict[str, str], fields: tuple[CredentialField, ...]) -> None:
+    missing = [field.label for field in fields if not credentials.get(field.name, "").strip()]
+    if missing:
+        raise ExternalAuthError("请填写" + "、".join(missing))
+
+
+def _nested_value(payload: Any, path: str) -> Any:
+    current = payload
+    for part in path.split("."):
+        if not isinstance(current, dict):
+            return None
+        current = current.get(part)
+    return current
+
+
+def _auth_header(config: dict[str, Any], token: str, *, default_name: str, default_template: str) -> dict[str, str]:
+    name = str(config.get("auth_header_name") or default_name).strip()
+    template = str(config.get("auth_header_template") or default_template)
+    if any(char in name for char in "\r\n:") or any(char in template for char in "\r\n"):
+        raise ExternalAuthError("认证 Header 配置非法")
+    if "{token}" not in template:
+        raise ExternalAuthError("认证 Header 模板必须包含 {token}")
+    return {name: template.replace("{token}", token)}
+
+
+class PasswordJwtAuth:
+    credential_fields = (
+        CredentialField("username", "用户名", secret=False, autocomplete="username"),
+        CredentialField("password", "密码", autocomplete="current-password"),
+    )
+
+    def headers(self, *, client, base_url, credentials, config) -> dict[str, str]:
+        _required(credentials, self.credential_fields)
+        login_path = str(config.get("login_path") or "/api/auth/token/").strip()
+        if not login_path.startswith("/") or "://" in login_path:
+            raise ExternalAuthError("login_path 必须是站内绝对路径")
+        username_field = str(config.get("username_field") or "username").strip()
+        password_field = str(config.get("password_field") or "password").strip()
+        token_field = str(config.get("token_field") or "access").strip()
+        try:
+            response = client.post(
+                urljoin(base_url + "/", login_path.lstrip("/")),
+                json={
+                    username_field: credentials["username"],
+                    password_field: credentials["password"],
+                },
+            )
+        except httpx.HTTPError as exc:
+            raise ExternalAuthError(f"外部系统登录连接失败: {type(exc).__name__}") from exc
+        if response.status_code >= 400:
+            raise ExternalAuthError(f"外部系统登录失败(HTTP {response.status_code})")
+        try:
+            token = _nested_value(response.json(), token_field)
+        except ValueError:
+            token = None
+        if not isinstance(token, str) or not token:
+            raise ExternalAuthError(f"外部系统登录响应缺少 {token_field}")
+        return _auth_header(
+            config, token, default_name="Authorization", default_template="Bearer {token}"
+        )
+
+
+class ApiKeyAuth:
+    credential_fields = (CredentialField("api_key", "API Key"),)
+
+    def headers(self, *, client, base_url, credentials, config) -> dict[str, str]:
+        _required(credentials, self.credential_fields)
+        return _auth_header(
+            config, credentials["api_key"], default_name="X-API-Key", default_template="{token}"
+        )
+
+
+class BearerTokenAuth:
+    credential_fields = (CredentialField("token", "Bearer Token"),)
+
+    def headers(self, *, client, base_url, credentials, config) -> dict[str, str]:
+        _required(credentials, self.credential_fields)
+        return _auth_header(
+            config, credentials["token"], default_name="Authorization", default_template="Bearer {token}"
+        )
+
+
+_AUTH_STRATEGIES: dict[str, AuthStrategy] = {
+    "password_jwt": PasswordJwtAuth(),
+    "api_key": ApiKeyAuth(),
+    "bearer_token": BearerTokenAuth(),
+}
+
+
+def get_auth_strategy(auth_type: str) -> AuthStrategy:
+    strategy = _AUTH_STRATEGIES.get((auth_type or "").strip())
+    if strategy is None:
+        raise ExternalAuthError(f"不支持的认证方式: {auth_type}")
+    return strategy
+
+
+def auth_catalog() -> list[dict[str, Any]]:
+    titles = {
+        "password_jwt": "用户名密码换取 Token",
+        "api_key": "API Key",
+        "bearer_token": "Bearer Token",
+    }
+    return [
+        {
+            "auth_type": key,
+            "title": titles[key],
+            "credential_fields": [field.as_dict() for field in strategy.credential_fields],
+        }
+        for key, strategy in _AUTH_STRATEGIES.items()
+    ]

core/external_systems/factory.py

@@ -1,419 +1,28 @@
-"""Factory MES OpenAPI connector。
+"""Factory MES 兼容入口。
 
-目标地址全部来自管理员维护的可信系统目录；模型和普通用户只能传 operation_id
+新代码使用 :mod:`core.external_systems.openapi`；保留原类名，避免已有测试和内部引用
-与结构化参数，不能传 URL。
+在通用化过程中发生无意义破坏。
 """
 from __future__ import annotations
 
-import json
+from typing import Any
-import re
-import time
-from dataclasses import dataclass
-from threading import Lock
-from typing import Any, Optional
-from urllib.parse import quote, urljoin, urlparse
 
-import httpx
+from .openapi import OpenApiClient, OpenApiConfig, OpenApiError, _SPEC_CACHE
+from .registry import merged_config
 
 
-class FactoryMesError(RuntimeError):
+FactoryMesError = OpenApiError
-    pass
 
 
-_HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete")
+class FactoryMesConfig(OpenApiConfig):
-_SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
-_SPEC_LOCK = Lock()
-
-DEFAULT_QUERY_GUIDANCE = (
-    "产量、良率、缺陷、库存、绩效、趋势和按日/月汇总等统计聚合查询，"
-    "统一先调用 BI dataset list，再执行匹配的数据集。日志和业务明细列表用于"
-    "用户明确要求查看逐条记录、编号或追溯过程的场景。未匹配到 dataset 时，"
-    "先限定范围或向用户确认明细查询需求。"
-)
-DEFAULT_RECOMMENDED_OPERATIONS = ("bi_dataset_list", "bi_dataset_exec")
-
-
-def _bool_value(value: Any, default: bool) -> bool:
-    raw = str(value if value is not None else "").strip().lower()
-    if not raw:
-        return default
-    return raw in {"1", "true", "yes", "on"}
-
-
-def _validated_http_url(raw: str, label: str) -> str:
-    value = (raw or "").strip().rstrip("/")
-    parsed = urlparse(value)
-    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
-        raise FactoryMesError(f"{label} 必须是有效的 http(s) URL")
-    if parsed.username or parsed.password:
-        raise FactoryMesError(f"{label} 不能内嵌凭据")
-    return value
-
-
-@dataclass(frozen=True)
-class FactoryMesConfig:
-    base_url: str
-    openapi_url: str
-    login_path: str
-    allowed_post_operations: frozenset[str]
-    timeout_seconds: float
-    max_result_bytes: int
-    max_total_result_bytes: int
-    max_page_size: int
-    verify_tls: bool
-    query_guidance: str
-    recommended_operation_ids: tuple[str, ...]
-
     @classmethod
     def from_mapping(cls, data: dict[str, Any]) -> "FactoryMesConfig":
-        """从管理员保存的可信目录配置构建运行态配置。"""
+        common = OpenApiConfig.from_mapping(merged_config("factory_mes", data))
-        base = _validated_http_url(str(data.get("base_url") or ""), "base_url")
+        return cls(**common.__dict__)
-        spec = _validated_http_url(
-            str(data.get("openapi_url") or ""), "openapi_url"
-        )
-        login_path = str(data.get("login_path") or "/api/auth/token/").strip()
-        if not login_path.startswith("/") or "://" in login_path:
-            raise FactoryMesError("login_path 必须是站内绝对路径")
-        raw_allowed = data.get("allowed_post_operations") or []
-        if isinstance(raw_allowed, str):
-            raw_allowed = raw_allowed.split(",")
-        if not isinstance(raw_allowed, (list, tuple, set)):
-            raise FactoryMesError("allowed_post_operations 必须是字符串数组")
-        allowed = frozenset(str(item).strip() for item in raw_allowed if str(item).strip())
-        guidance = str(data.get("query_guidance") or DEFAULT_QUERY_GUIDANCE).strip()
-        if len(guidance) > 4000:
-            raise FactoryMesError("query_guidance 不能超过 4000 字符")
-        raw_recommended = data.get(
-            "recommended_operation_ids", DEFAULT_RECOMMENDED_OPERATIONS
-        )
-        if isinstance(raw_recommended, str):
-            raw_recommended = raw_recommended.split(",")
-        if not isinstance(raw_recommended, (list, tuple, set)):
-            raise FactoryMesError("recommended_operation_ids 必须是字符串数组")
-        recommended = tuple(dict.fromkeys(
-            str(item).strip() for item in raw_recommended if str(item).strip()
-        ))
-        if len(recommended) > 30 or any(len(item) > 200 for item in recommended):
-            raise FactoryMesError("recommended_operation_ids 最多 30 项且每项不超过 200 字符")
-        max_result = max(4096, min(int(data.get("max_result_bytes", 65536)), 1048576))
-        return cls(
-            base_url=base,
-            openapi_url=spec,
-            login_path=login_path,
-            allowed_post_operations=allowed,
-            timeout_seconds=max(1.0, min(float(data.get("timeout_seconds", 15)), 60.0)),
-            max_result_bytes=max_result,
-            max_total_result_bytes=max(
-                max_result,
-                min(int(data.get("max_total_result_bytes", 262144)), 4194304),
-            ),
-            max_page_size=max(1, min(int(data.get("max_page_size", 200)), 1000)),
-            verify_tls=_bool_value(data.get("verify_tls"), True),
-            query_guidance=guidance,
-            recommended_operation_ids=recommended,
-        )
 
 
-class FactoryMesClient:
+class FactoryMesClient(OpenApiClient):
     def __init__(self, username: str, password: str, cfg: FactoryMesConfig):
         self.username = username
         self.password = password
-        self.cfg = cfg
+        super().__init__({"username": username, "password": password}, cfg)
-
-    def _client(self) -> httpx.Client:
-        return httpx.Client(
-            timeout=self.cfg.timeout_seconds,
-            verify=self.cfg.verify_tls,
-            follow_redirects=False,
-        )
-
-    def authenticate(self) -> str:
-        url = urljoin(self.cfg.base_url + "/", self.cfg.login_path.lstrip("/"))
-        try:
-            with self._client() as client:
-                response = client.post(
-                    url,
-                    json={"username": self.username, "password": self.password},
-                )
-        except httpx.HTTPError as exc:
-            raise FactoryMesError(f"Factory MES 登录连接失败: {type(exc).__name__}") from exc
-        if response.status_code >= 400:
-            raise FactoryMesError(f"Factory MES 登录失败(HTTP {response.status_code})")
-        try:
-            token = response.json().get("access", "")
-        except (ValueError, AttributeError):
-            token = ""
-        if not isinstance(token, str) or not token:
-            raise FactoryMesError("Factory MES 登录响应缺少 access token")
-        return token
-
-    def _fetch_spec(self, token: str) -> dict[str, Any]:
-        now = time.monotonic()
-        with _SPEC_LOCK:
-            hit = _SPEC_CACHE.get(self.cfg.openapi_url)
-            if hit and now - hit[0] < 300:
-                return hit[1]
-        try:
-            with self._client() as client:
-                response = client.get(
-                    self.cfg.openapi_url,
-                    headers={"Authorization": f"Bearer {token}"},
-                )
-        except httpx.HTTPError as exc:
-            raise FactoryMesError(f"Factory OpenAPI 获取失败: {type(exc).__name__}") from exc
-        if response.status_code >= 400:
-            raise FactoryMesError(f"Factory OpenAPI 获取失败(HTTP {response.status_code})")
-        try:
-            spec = response.json()
-        except ValueError as exc:
-            raise FactoryMesError("Factory OpenAPI 不是有效 JSON") from exc
-        if not isinstance(spec, dict) or not isinstance(spec.get("paths"), dict):
-            raise FactoryMesError("Factory OpenAPI 缺少 paths")
-        with _SPEC_LOCK:
-            _SPEC_CACHE[self.cfg.openapi_url] = (now, spec)
-        return spec
-
-    @staticmethod
-    def _operation_id(method: str, path: str, operation: dict[str, Any]) -> str:
-        explicit = operation.get("operationId")
-        if isinstance(explicit, str) and explicit.strip():
-            return explicit.strip()
-        safe_path = re.sub(r"[^a-zA-Z0-9]+", "_", path).strip("_")
-        return f"{method}_{safe_path}"
-
-    @classmethod
-    def _operations(cls, spec: dict[str, Any]) -> list[dict[str, Any]]:
-        results: list[dict[str, Any]] = []
-        for path, path_item in (spec.get("paths") or {}).items():
-            if not isinstance(path_item, dict):
-                continue
-            common = path_item.get("parameters") or []
-            for method in _HTTP_METHODS:
-                operation = path_item.get(method)
-                if not isinstance(operation, dict):
-                    continue
-                params = list(common) + list(operation.get("parameters") or [])
-                results.append({
-                    "operation_id": cls._operation_id(method, path, operation),
-                    "method": method.upper(),
-                    "path": path,
-                    "summary": operation.get("summary") or "",
-                    "description": operation.get("description") or "",
-                    "tags": operation.get("tags") or [],
-                    "parameters": params,
-                    "request_body": operation.get("requestBody"),
-                })
-        return results
-
-    def _spec_base_path(self, spec: dict[str, Any]) -> str:
-        """Return the API path prefix declared by Swagger 2 / OpenAPI 3.
-
-        The remote specification may describe a different host, but an external
-        system definition is the only authority allowed to choose the target
-        origin.  OpenAPI ``servers`` therefore contributes only a same-origin
-        path prefix.
-        """
-        raw_base_path = spec.get("basePath")
-        if raw_base_path is not None:
-            if not isinstance(raw_base_path, str) or not raw_base_path.startswith("/"):
-                raise FactoryMesError("Swagger basePath 必须是站内绝对路径")
-            parsed = urlparse(raw_base_path)
-            if parsed.netloc or parsed.query or parsed.fragment or "://" in raw_base_path:
-                raise FactoryMesError("Swagger basePath 非法")
-            return parsed.path.rstrip("/")
-
-        servers = spec.get("servers")
-        if not isinstance(servers, list) or not servers:
-            return ""
-        server = servers[0]
-        raw_url = server.get("url") if isinstance(server, dict) else None
-        if not isinstance(raw_url, str) or not raw_url.strip():
-            raise FactoryMesError("OpenAPI server URL 无效")
-        raw_url = raw_url.strip()
-        if "{" in raw_url or "}" in raw_url:
-            raise FactoryMesError("OpenAPI server URL 包含未解析变量")
-        declared = urlparse(raw_url)
-        base = urlparse(self.cfg.base_url)
-        if declared.netloc and (declared.scheme, declared.netloc) != (
-            base.scheme,
-            base.netloc,
-        ):
-            raise FactoryMesError("OpenAPI server 越出 Factory MES 主机")
-        if declared.query or declared.fragment:
-            raise FactoryMesError("OpenAPI server URL 不能包含查询或片段")
-        return ("/" + declared.path.lstrip("/")).rstrip("/")
-
-    def _operation_url(self, spec: dict[str, Any], operation_path: str) -> str:
-        prefix = self._spec_base_path(spec)
-        base = urlparse(self.cfg.base_url)
-        configured_path = base.path.rstrip("/")
-        path = operation_path
-        configured_has_prefix = bool(prefix) and (
-            configured_path == prefix or configured_path.endswith(prefix)
-        )
-        operation_has_prefix = bool(prefix) and (
-            path == prefix or path.startswith(prefix + "/")
-        )
-        if configured_has_prefix and operation_has_prefix:
-            path = path[len(prefix):] or "/"
-        elif prefix and not configured_has_prefix and not operation_has_prefix:
-            path = prefix + "/" + path.lstrip("/")
-        combined_path = "/".join(
-            part.strip("/") for part in (configured_path, path) if part.strip("/")
-        )
-        if operation_path.endswith("/") and combined_path:
-            combined_path += "/"
-        url = urljoin(f"{base.scheme}://{base.netloc}/", combined_path)
-        call_origin = urlparse(url)
-        if (call_origin.scheme, call_origin.netloc) != (base.scheme, base.netloc):
-            raise FactoryMesError("接口目标越出 Factory MES 主机")
-        return url
-
-    def test_connection(self) -> dict[str, Any]:
-        token = self.authenticate()
-        spec = self._fetch_spec(token)
-        return {"operation_count": len(self._operations(spec))}
-
-    def search(self, query: str, limit: int = 12) -> list[dict[str, Any]]:
-        query = (query or "").strip().lower()
-        if not query:
-            raise FactoryMesError("query 不能为空")
-        token = self.authenticate()
-        spec = self._fetch_spec(token)
-        terms = list(dict.fromkeys(
-            [query] + [x for x in re.split(r"[\s,，。/]+", query) if len(x) >= 2]
-        ))
-        recommended_order = {
-            operation_id: index
-            for index, operation_id in enumerate(self.cfg.recommended_operation_ids)
-        }
-        scored: list[tuple[int, int, dict[str, Any]]] = []
-        for op in self._operations(spec):
-            method = op["method"].lower()
-            if method not in {"get", "head"} and not (
-                method == "post"
-                and op["operation_id"] in self.cfg.allowed_post_operations
-            ):
-                continue
-            hay = " ".join([
-                op["operation_id"], op["path"], op["summary"], op["description"],
-                " ".join(str(x) for x in op["tags"]),
-            ]).lower()
-            score = sum(5 if term == query and term in hay else 1 for term in terms if term in hay)
-            recommended = op["operation_id"] in recommended_order
-            if score or recommended:
-                compact = dict(op)
-                compact["parameters"] = [
-                    {
-                        "name": p.get("name"),
-                        "in": p.get("in"),
-                        "required": bool(p.get("required")),
-                        "type": p.get("type") or (p.get("schema") or {}).get("type"),
-                        "description": p.get("description") or "",
-                    }
-                    for p in op["parameters"] if isinstance(p, dict) and "$ref" not in p
-                ]
-                compact.pop("request_body", None)
-                compact["recommended"] = recommended
-                scored.append((
-                    0 if recommended else 1,
-                    recommended_order.get(op["operation_id"], -score),
-                    compact,
-                ))
-        scored.sort(key=lambda item: (item[0], item[1], item[2]["operation_id"]))
-        return [item[2] for item in scored[: max(1, min(int(limit), 30))]]
-
-    def call(
-        self,
-        operation_id: str,
-        arguments: Optional[dict[str, Any]] = None,
-        body: Any = None,
-    ) -> dict[str, Any]:
-        token = self.authenticate()
-        spec = self._fetch_spec(token)
-        matches = [op for op in self._operations(spec) if op["operation_id"] == operation_id]
-        if len(matches) != 1:
-            raise FactoryMesError("operation_id 不存在或不唯一，请先搜索接口")
-        op = matches[0]
-        if not op["path"].startswith("/") or "://" in op["path"]:
-            raise FactoryMesError("OpenAPI operation path 非法")
-        method = op["method"].lower()
-        if method not in {"get", "head"} and not (
-            method == "post" and operation_id in self.cfg.allowed_post_operations
-        ):
-            raise FactoryMesError(f"operation {operation_id} 未列入只读调用范围")
-
-        supplied = dict(arguments or {})
-        path = op["path"]
-        query: dict[str, Any] = {}
-        headers = {"Authorization": f"Bearer {token}"}
-        request_body = body
-        for param in op["parameters"]:
-            if not isinstance(param, dict) or "$ref" in param:
-                continue
-            name = param.get("name")
-            location = param.get("in")
-            if not isinstance(name, str):
-                continue
-            # Swagger 2 的 body 参数既可按搜索结果中的参数名放在 arguments，
-            # 也可使用元工具独立的 body 字段；两者只取一个。
-            present = name in supplied or (location == "body" and request_body is not None)
-            if param.get("required") and not present:
-                raise FactoryMesError(f"缺少必填参数: {name}")
-            if name not in supplied:
-                continue
-            value = supplied.pop(name)
-            if location == "path":
-                path = path.replace("{" + name + "}", quote(str(value), safe=""))
-            elif location == "query":
-                if name == "page_size":
-                    try:
-                        value = max(1, min(int(value), self.cfg.max_page_size))
-                    except (TypeError, ValueError) as exc:
-                        raise FactoryMesError("page_size 必须是整数") from exc
-                elif name == "page" and str(value).strip() == "0":
-                    raise FactoryMesError(
-                        "外部系统查询不允许 page=0 关闭分页，请使用 dataset 或分页查看明细"
-                    )
-                elif name == "pageoff" and _bool_value(value, False):
-                    raise FactoryMesError(
-                        "外部系统查询不允许关闭分页，请使用 dataset 或分页查看明细"
-                    )
-                query[name] = value
-            elif location == "body" and request_body is None:
-                request_body = value
-        if supplied:
-            raise FactoryMesError("存在接口定义之外的参数: " + ", ".join(sorted(supplied)))
-        if "{" in path or "}" in path:
-            raise FactoryMesError("路径参数未完整提供")
-
-        url = self._operation_url(spec, path)
-        try:
-            with self._client() as client:
-                response = client.request(
-                    method.upper(),
-                    url,
-                    params=query,
-                    json=request_body if method == "post" else None,
-                    headers=headers,
-                )
-        except httpx.HTTPError as exc:
-            raise FactoryMesError(f"Factory 接口调用失败: {type(exc).__name__}") from exc
-        if response.status_code >= 400:
-            raise FactoryMesError(f"Factory 接口返回 HTTP {response.status_code}")
-        content_type = response.headers.get("content-type", "")
-        try:
-            payload: Any = response.json() if "json" in content_type else response.text
-        except ValueError:
-            payload = response.text
-        encoded = json.dumps(payload, ensure_ascii=False, default=str)
-        truncated = len(encoded.encode("utf-8")) > self.cfg.max_result_bytes
-        if truncated:
-            encoded = encoded.encode("utf-8")[: self.cfg.max_result_bytes].decode("utf-8", "ignore")
-            payload = encoded
-        return {
-            "operation_id": operation_id,
-            "status_code": response.status_code,
-            "truncated": truncated,
-            "data": payload,
-        }

core/external_systems/openapi.py

@@ -0,0 +1,430 @@
+"""通用 OpenAPI 外部系统连接器。
+
+目标地址全部来自管理员维护的可信系统目录；模型和普通用户只能传 operation_id
+与结构化参数，不能传 URL。
+"""
+from __future__ import annotations
+
+import json
+import hashlib
+import re
+import time
+from dataclasses import dataclass, field
+from threading import Lock
+from typing import Any, Optional
+from urllib.parse import quote, urljoin, urlparse
+
+import httpx
+
+from .auth import ExternalAuthError, get_auth_strategy
+
+
+class OpenApiError(RuntimeError):
+    pass
+
+
+_HTTP_METHODS = ("get", "head", "post", "put", "patch", "delete")
+_SPEC_CACHE: dict[str, tuple[float, dict[str, Any]]] = {}
+_SPEC_LOCK = Lock()
+
+def _bool_value(value: Any, default: bool) -> bool:
+    raw = str(value if value is not None else "").strip().lower()
+    if not raw:
+        return default
+    return raw in {"1", "true", "yes", "on"}
+
+
+def _validated_http_url(raw: str, label: str) -> str:
+    value = (raw or "").strip().rstrip("/")
+    parsed = urlparse(value)
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise OpenApiError(f"{label} 必须是有效的 http(s) URL")
+    if parsed.username or parsed.password:
+        raise OpenApiError(f"{label} 不能内嵌凭据")
+    return value
+
+
+@dataclass(frozen=True)
+class OpenApiConfig:
+    base_url: str
+    openapi_url: str
+    login_path: str
+    allowed_post_operations: frozenset[str]
+    timeout_seconds: float
+    max_result_bytes: int
+    max_total_result_bytes: int
+    max_page_size: int
+    verify_tls: bool
+    query_guidance: str
+    recommended_operation_ids: tuple[str, ...]
+    auth_type: str = "password_jwt"
+    auth_config: dict[str, Any] = field(default_factory=lambda: {
+        "login_path": "/api/auth/token/",
+        "username_field": "username",
+        "password_field": "password",
+        "token_field": "access",
+        "auth_header_name": "Authorization",
+        "auth_header_template": "Bearer {token}",
+    })
+
+    @classmethod
+    def from_mapping(cls, data: dict[str, Any]) -> "OpenApiConfig":
+        """从管理员保存的可信目录配置构建运行态配置。"""
+        base = _validated_http_url(str(data.get("base_url") or ""), "base_url")
+        spec = _validated_http_url(
+            str(data.get("openapi_url") or ""), "openapi_url"
+        )
+        login_path = str(data.get("login_path") or "/api/auth/token/").strip()
+        if not login_path.startswith("/") or "://" in login_path:
+            raise OpenApiError("login_path 必须是站内绝对路径")
+        raw_allowed = data.get("allowed_post_operations") or []
+        if isinstance(raw_allowed, str):
+            raw_allowed = raw_allowed.split(",")
+        if not isinstance(raw_allowed, (list, tuple, set)):
+            raise OpenApiError("allowed_post_operations 必须是字符串数组")
+        allowed = frozenset(str(item).strip() for item in raw_allowed if str(item).strip())
+        guidance = str(data.get("query_guidance") or "").strip()
+        if len(guidance) > 4000:
+            raise OpenApiError("query_guidance 不能超过 4000 字符")
+        raw_recommended = data.get(
+            "recommended_operation_ids", []
+        )
+        if isinstance(raw_recommended, str):
+            raw_recommended = raw_recommended.split(",")
+        if not isinstance(raw_recommended, (list, tuple, set)):
+            raise OpenApiError("recommended_operation_ids 必须是字符串数组")
+        recommended = tuple(dict.fromkeys(
+            str(item).strip() for item in raw_recommended if str(item).strip()
+        ))
+        if len(recommended) > 30 or any(len(item) > 200 for item in recommended):
+            raise OpenApiError("recommended_operation_ids 最多 30 项且每项不超过 200 字符")
+        max_result = max(4096, min(int(data.get("max_result_bytes", 65536)), 1048576))
+        return cls(
+            base_url=base,
+            openapi_url=spec,
+            login_path=login_path,
+            allowed_post_operations=allowed,
+            timeout_seconds=max(1.0, min(float(data.get("timeout_seconds", 15)), 60.0)),
+            max_result_bytes=max_result,
+            max_total_result_bytes=max(
+                max_result,
+                min(int(data.get("max_total_result_bytes", 262144)), 4194304),
+            ),
+            max_page_size=max(1, min(int(data.get("max_page_size", 200)), 1000)),
+            verify_tls=_bool_value(data.get("verify_tls"), True),
+            query_guidance=guidance,
+            recommended_operation_ids=recommended,
+            auth_type=str(data.get("auth_type") or "password_jwt").strip(),
+            auth_config={
+                key: data[key]
+                for key in (
+                    "login_path", "username_field", "password_field", "token_field",
+                    "auth_header_name", "auth_header_template",
+                )
+                if key in data
+            },
+        )
+
+
+class OpenApiClient:
+    def __init__(
+        self,
+        credentials: dict[str, str],
+        cfg: OpenApiConfig,
+        *,
+        cache_namespace: str = "",
+    ):
+        self.credentials = credentials
+        self.cfg = cfg
+        identity = cache_namespace or json.dumps(credentials, sort_keys=True, ensure_ascii=False)
+        digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()
+        self._spec_cache_key = f"{cfg.openapi_url}:{digest}"
+
+    def _client(self) -> httpx.Client:
+        return httpx.Client(
+            timeout=self.cfg.timeout_seconds,
+            verify=self.cfg.verify_tls,
+            follow_redirects=False,
+        )
+
+    def authenticate(self) -> dict[str, str]:
+        try:
+            with self._client() as client:
+                return get_auth_strategy(self.cfg.auth_type).headers(
+                    client=client,
+                    base_url=self.cfg.base_url,
+                    credentials=self.credentials,
+                    config=self.cfg.auth_config,
+                )
+        except ExternalAuthError as exc:
+            raise OpenApiError(str(exc)) from exc
+
+    def _fetch_spec(self, headers: dict[str, str]) -> dict[str, Any]:
+        now = time.monotonic()
+        with _SPEC_LOCK:
+            hit = _SPEC_CACHE.get(self._spec_cache_key)
+            if hit and now - hit[0] < 300:
+                return hit[1]
+        try:
+            with self._client() as client:
+                response = client.get(
+                    self.cfg.openapi_url,
+                    headers=headers,
+                )
+        except httpx.HTTPError as exc:
+            raise OpenApiError(f"OpenAPI 获取失败: {type(exc).__name__}") from exc
+        if response.status_code >= 400:
+            raise OpenApiError(f"OpenAPI 获取失败(HTTP {response.status_code})")
+        try:
+            spec = response.json()
+        except ValueError as exc:
+            raise OpenApiError("OpenAPI 文档不是有效 JSON") from exc
+        if not isinstance(spec, dict) or not isinstance(spec.get("paths"), dict):
+            raise OpenApiError("OpenAPI 文档缺少 paths")
+        with _SPEC_LOCK:
+            _SPEC_CACHE[self._spec_cache_key] = (now, spec)
+        return spec
+
+    @staticmethod
+    def _operation_id(method: str, path: str, operation: dict[str, Any]) -> str:
+        explicit = operation.get("operationId")
+        if isinstance(explicit, str) and explicit.strip():
+            return explicit.strip()
+        safe_path = re.sub(r"[^a-zA-Z0-9]+", "_", path).strip("_")
+        return f"{method}_{safe_path}"
+
+    @classmethod
+    def _operations(cls, spec: dict[str, Any]) -> list[dict[str, Any]]:
+        results: list[dict[str, Any]] = []
+        for path, path_item in (spec.get("paths") or {}).items():
+            if not isinstance(path_item, dict):
+                continue
+            common = path_item.get("parameters") or []
+            for method in _HTTP_METHODS:
+                operation = path_item.get(method)
+                if not isinstance(operation, dict):
+                    continue
+                params = list(common) + list(operation.get("parameters") or [])
+                results.append({
+                    "operation_id": cls._operation_id(method, path, operation),
+                    "method": method.upper(),
+                    "path": path,
+                    "summary": operation.get("summary") or "",
+                    "description": operation.get("description") or "",
+                    "tags": operation.get("tags") or [],
+                    "parameters": params,
+                    "request_body": operation.get("requestBody"),
+                })
+        return results
+
+    def _spec_base_path(self, spec: dict[str, Any]) -> str:
+        """Return the API path prefix declared by Swagger 2 / OpenAPI 3.
+
+        The remote specification may describe a different host, but an external
+        system definition is the only authority allowed to choose the target
+        origin.  OpenAPI ``servers`` therefore contributes only a same-origin
+        path prefix.
+        """
+        raw_base_path = spec.get("basePath")
+        if raw_base_path is not None:
+            if not isinstance(raw_base_path, str) or not raw_base_path.startswith("/"):
+                raise OpenApiError("Swagger basePath 必须是站内绝对路径")
+            parsed = urlparse(raw_base_path)
+            if parsed.netloc or parsed.query or parsed.fragment or "://" in raw_base_path:
+                raise OpenApiError("Swagger basePath 非法")
+            return parsed.path.rstrip("/")
+
+        servers = spec.get("servers")
+        if not isinstance(servers, list) or not servers:
+            return ""
+        server = servers[0]
+        raw_url = server.get("url") if isinstance(server, dict) else None
+        if not isinstance(raw_url, str) or not raw_url.strip():
+            raise OpenApiError("OpenAPI server URL 无效")
+        raw_url = raw_url.strip()
+        if "{" in raw_url or "}" in raw_url:
+            raise OpenApiError("OpenAPI server URL 包含未解析变量")
+        declared = urlparse(raw_url)
+        base = urlparse(self.cfg.base_url)
+        if declared.netloc and (declared.scheme, declared.netloc) != (
+            base.scheme,
+            base.netloc,
+        ):
+            raise OpenApiError("OpenAPI server 越出 Factory MES 主机")
+        if declared.query or declared.fragment:
+            raise OpenApiError("OpenAPI server URL 不能包含查询或片段")
+        return ("/" + declared.path.lstrip("/")).rstrip("/")
+
+    def _operation_url(self, spec: dict[str, Any], operation_path: str) -> str:
+        prefix = self._spec_base_path(spec)
+        base = urlparse(self.cfg.base_url)
+        configured_path = base.path.rstrip("/")
+        path = operation_path
+        configured_has_prefix = bool(prefix) and (
+            configured_path == prefix or configured_path.endswith(prefix)
+        )
+        operation_has_prefix = bool(prefix) and (
+            path == prefix or path.startswith(prefix + "/")
+        )
+        if configured_has_prefix and operation_has_prefix:
+            path = path[len(prefix):] or "/"
+        elif prefix and not configured_has_prefix and not operation_has_prefix:
+            path = prefix + "/" + path.lstrip("/")
+        combined_path = "/".join(
+            part.strip("/") for part in (configured_path, path) if part.strip("/")
+        )
+        if operation_path.endswith("/") and combined_path:
+            combined_path += "/"
+        url = urljoin(f"{base.scheme}://{base.netloc}/", combined_path)
+        call_origin = urlparse(url)
+        if (call_origin.scheme, call_origin.netloc) != (base.scheme, base.netloc):
+            raise OpenApiError("接口目标越出管理员配置的主机")
+        return url
+
+    def test_connection(self) -> dict[str, Any]:
+        headers = self.authenticate()
+        spec = self._fetch_spec(headers)
+        return {"operation_count": len(self._operations(spec))}
+
+    def search(self, query: str, limit: int = 12) -> list[dict[str, Any]]:
+        query = (query or "").strip().lower()
+        if not query:
+            raise OpenApiError("query 不能为空")
+        headers = self.authenticate()
+        spec = self._fetch_spec(headers)
+        terms = list(dict.fromkeys(
+            [query] + [x for x in re.split(r"[\s,，。/]+", query) if len(x) >= 2]
+        ))
+        recommended_order = {
+            operation_id: index
+            for index, operation_id in enumerate(self.cfg.recommended_operation_ids)
+        }
+        scored: list[tuple[int, int, dict[str, Any]]] = []
+        for op in self._operations(spec):
+            method = op["method"].lower()
+            if method not in {"get", "head"} and not (
+                method == "post"
+                and op["operation_id"] in self.cfg.allowed_post_operations
+            ):
+                continue
+            hay = " ".join([
+                op["operation_id"], op["path"], op["summary"], op["description"],
+                " ".join(str(x) for x in op["tags"]),
+            ]).lower()
+            score = sum(5 if term == query and term in hay else 1 for term in terms if term in hay)
+            recommended = op["operation_id"] in recommended_order
+            if score or recommended:
+                compact = dict(op)
+                compact["parameters"] = [
+                    {
+                        "name": p.get("name"),
+                        "in": p.get("in"),
+                        "required": bool(p.get("required")),
+                        "type": p.get("type") or (p.get("schema") or {}).get("type"),
+                        "description": p.get("description") or "",
+                    }
+                    for p in op["parameters"] if isinstance(p, dict) and "$ref" not in p
+                ]
+                compact.pop("request_body", None)
+                compact["recommended"] = recommended
+                scored.append((
+                    0 if recommended else 1,
+                    recommended_order.get(op["operation_id"], -score),
+                    compact,
+                ))
+        scored.sort(key=lambda item: (item[0], item[1], item[2]["operation_id"]))
+        return [item[2] for item in scored[: max(1, min(int(limit), 30))]]
+
+    def call(
+        self,
+        operation_id: str,
+        arguments: Optional[dict[str, Any]] = None,
+        body: Any = None,
+    ) -> dict[str, Any]:
+        headers = self.authenticate()
+        spec = self._fetch_spec(headers)
+        matches = [op for op in self._operations(spec) if op["operation_id"] == operation_id]
+        if len(matches) != 1:
+            raise OpenApiError("operation_id 不存在或不唯一，请先搜索接口")
+        op = matches[0]
+        if not op["path"].startswith("/") or "://" in op["path"]:
+            raise OpenApiError("OpenAPI operation path 非法")
+        method = op["method"].lower()
+        if method not in {"get", "head"} and not (
+            method == "post" and operation_id in self.cfg.allowed_post_operations
+        ):
+            raise OpenApiError(f"operation {operation_id} 未列入只读调用范围")
+
+        supplied = dict(arguments or {})
+        path = op["path"]
+        query: dict[str, Any] = {}
+        request_body = body
+        for param in op["parameters"]:
+            if not isinstance(param, dict) or "$ref" in param:
+                continue
+            name = param.get("name")
+            location = param.get("in")
+            if not isinstance(name, str):
+                continue
+            # Swagger 2 的 body 参数既可按搜索结果中的参数名放在 arguments，
+            # 也可使用元工具独立的 body 字段；两者只取一个。
+            present = name in supplied or (location == "body" and request_body is not None)
+            if param.get("required") and not present:
+                raise OpenApiError(f"缺少必填参数: {name}")
+            if name not in supplied:
+                continue
+            value = supplied.pop(name)
+            if location == "path":
+                path = path.replace("{" + name + "}", quote(str(value), safe=""))
+            elif location == "query":
+                if name == "page_size":
+                    try:
+                        value = max(1, min(int(value), self.cfg.max_page_size))
+                    except (TypeError, ValueError) as exc:
+                        raise OpenApiError("page_size 必须是整数") from exc
+                elif name == "page" and str(value).strip() == "0":
+                    raise OpenApiError(
+                        "外部系统查询不允许 page=0 关闭分页，请使用 dataset 或分页查看明细"
+                    )
+                elif name == "pageoff" and _bool_value(value, False):
+                    raise OpenApiError(
+                        "外部系统查询不允许关闭分页，请使用 dataset 或分页查看明细"
+                    )
+                query[name] = value
+            elif location == "body" and request_body is None:
+                request_body = value
+        if supplied:
+            raise OpenApiError("存在接口定义之外的参数: " + ", ".join(sorted(supplied)))
+        if "{" in path or "}" in path:
+            raise OpenApiError("路径参数未完整提供")
+
+        url = self._operation_url(spec, path)
+        try:
+            with self._client() as client:
+                response = client.request(
+                    method.upper(),
+                    url,
+                    params=query,
+                    json=request_body if method == "post" else None,
+                    headers=headers,
+                )
+        except httpx.HTTPError as exc:
+            raise OpenApiError(f"外部系统接口调用失败: {type(exc).__name__}") from exc
+        if response.status_code >= 400:
+            raise OpenApiError(f"外部系统接口返回 HTTP {response.status_code}")
+        content_type = response.headers.get("content-type", "")
+        try:
+            payload: Any = response.json() if "json" in content_type else response.text
+        except ValueError:
+            payload = response.text
+        encoded = json.dumps(payload, ensure_ascii=False, default=str)
+        truncated = len(encoded.encode("utf-8")) > self.cfg.max_result_bytes
+        if truncated:
+            encoded = encoded.encode("utf-8")[: self.cfg.max_result_bytes].decode("utf-8", "ignore")
+            payload = encoded
+        return {
+            "operation_id": operation_id,
+            "status_code": response.status_code,
+            "truncated": truncated,
+            "data": payload,
+        }

core/external_systems/registry.py

@@ -0,0 +1,97 @@
+"""外部系统 provider 注册表。
+
+标准 OpenAPI 系统通过数据库配置接入；只有非 OpenAPI 协议才需要新增 connector 文件。
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from .auth import ExternalAuthError, get_auth_strategy
+
+
+FACTORY_QUERY_GUIDANCE = (
+    "产量、良率、缺陷、库存、绩效、趋势和按日/月汇总等统计聚合查询，"
+    "统一先调用 BI dataset list，再执行匹配的数据集。日志和业务明细列表用于"
+    "用户明确要求查看逐条记录、编号或追溯过程的场景。未匹配到 dataset 时，"
+    "先限定范围或向用户确认明细查询需求。"
+)
+FACTORY_RECOMMENDED_OPERATIONS = ("bi_dataset_list", "bi_dataset_exec")
+
+
+@dataclass(frozen=True)
+class ProviderSpec:
+    provider: str
+    title: str
+    connector: str
+    default_auth_type: str
+    allowed_auth_types: tuple[str, ...]
+    defaults: dict[str, Any]
+
+
+_PROVIDERS = {
+    "factory_mes": ProviderSpec(
+        provider="factory_mes",
+        title="Factory MES",
+        connector="openapi",
+        default_auth_type="password_jwt",
+        allowed_auth_types=("password_jwt",),
+        defaults={
+            "login_path": "/api/auth/token/",
+            "username_field": "username",
+            "password_field": "password",
+            "token_field": "access",
+            "auth_header_name": "Authorization",
+            "auth_header_template": "Bearer {token}",
+            "query_guidance": FACTORY_QUERY_GUIDANCE,
+            "recommended_operation_ids": list(FACTORY_RECOMMENDED_OPERATIONS),
+        },
+    ),
+    "generic_openapi": ProviderSpec(
+        provider="generic_openapi",
+        title="通用 OpenAPI 系统",
+        connector="openapi",
+        default_auth_type="password_jwt",
+        allowed_auth_types=("password_jwt", "api_key", "bearer_token"),
+        defaults={
+            "login_path": "/api/auth/token/",
+            "username_field": "username",
+            "password_field": "password",
+            "token_field": "access",
+            "auth_header_name": "Authorization",
+            "auth_header_template": "Bearer {token}",
+            "query_guidance": "",
+            "recommended_operation_ids": [],
+        },
+    ),
+}
+
+
+def get_provider(provider: str) -> ProviderSpec:
+    result = _PROVIDERS.get((provider or "").strip())
+    if result is None:
+        raise ValueError(f"不支持的外部系统 provider: {provider}")
+    return result
+
+
+def provider_specs() -> tuple[ProviderSpec, ...]:
+    return tuple(_PROVIDERS.values())
+
+
+def merged_config(provider: str, config: dict[str, Any]) -> dict[str, Any]:
+    spec = get_provider(provider)
+    result = {**spec.defaults, **(config or {})}
+    auth_type = str(result.get("auth_type") or spec.default_auth_type).strip()
+    if auth_type not in spec.allowed_auth_types:
+        raise ValueError(f"{spec.title} 不支持认证方式 {auth_type}")
+    result["auth_type"] = auth_type
+    return result
+
+
+def credential_fields(provider: str, config: dict[str, Any]) -> list[dict[str, Any]]:
+    merged = merged_config(provider, config)
+    try:
+        strategy = get_auth_strategy(merged["auth_type"])
+    except ExternalAuthError as exc:
+        raise ValueError(str(exc)) from exc
+    return [field.as_dict() for field in strategy.credential_fields]

core/external_systems/service.py

@@ -14,22 +14,23 @@ from core.storage.models import ExternalSystem, ExternalSystemDefinition, User
 
 from .crypto import configured as crypto_configured
 from .crypto import decrypt_secret, encrypt_secret, mask_username
-from .factory import FactoryMesClient, FactoryMesConfig, FactoryMesError
+from .openapi import OpenApiClient, OpenApiConfig, OpenApiError
+from .registry import credential_fields, get_provider, merged_config, provider_specs
 
 
 class ExternalSystemError(RuntimeError):
     pass
 
 
-def _factory_config(data: dict[str, Any]) -> FactoryMesConfig:
+def _runtime_config(provider: str, data: dict[str, Any]) -> OpenApiConfig:
     try:
-        return FactoryMesConfig.from_mapping(data)
+        return OpenApiConfig.from_mapping(merged_config(provider, data))
-    except (FactoryMesError, TypeError, ValueError) as exc:
+    except (OpenApiError, TypeError, ValueError) as exc:
         raise ExternalSystemError(str(exc)) from exc
 
 
-def _normalized_config(data: dict[str, Any]) -> dict[str, Any]:
+def _normalized_config(provider: str, data: dict[str, Any]) -> dict[str, Any]:
-    cfg = _factory_config(data)
+    cfg = _runtime_config(provider, data)
     return {
         "base_url": cfg.base_url,
         "openapi_url": cfg.openapi_url,
@@ -42,6 +43,8 @@ def _normalized_config(data: dict[str, Any]) -> dict[str, Any]:
         "verify_tls": cfg.verify_tls,
         "query_guidance": cfg.query_guidance,
         "recommended_operation_ids": list(cfg.recommended_operation_ids),
+        "auth_type": cfg.auth_type,
+        **cfg.auth_config,
     }
 
 
@@ -50,12 +53,15 @@ def _definition_view(row: ExternalSystemDefinition, *, include_config: bool) ->
     result = {
         "definition_id": str(row.definition_id),
         "provider": row.provider,
+        "provider_title": get_provider(row.provider).title,
+        "connector": get_provider(row.provider).connector,
         "name": row.name,
         "enabled": row.enabled,
         "access_mode": row.access_mode,
         "host": urlparse(str(config.get("base_url") or "")).hostname or "",
         "created_at": row.created_at.isoformat() if row.created_at else None,
         "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+        "credential_fields": credential_fields(row.provider, config),
     }
     if include_config:
         result["config"] = config
@@ -106,7 +112,7 @@ def _sync_selected_users(
             user_id=uid,
             definition_id=definition.definition_id,
             provider=definition.provider,
-            connector="openapi",
+            connector=get_provider(definition.provider).connector,
             name=definition.name,
             credentials={},
             config={},
@@ -120,7 +126,6 @@ def provider_catalog(user_id: UUID) -> list[dict[str, Any]]:
             rows = s.execute(
                 select(ExternalSystemDefinition)
                 .where(
-                    ExternalSystemDefinition.provider == "factory_mes",
                     ExternalSystemDefinition.enabled.is_(True),
                     or_(
                         ExternalSystemDefinition.access_mode == "all",
@@ -135,18 +140,27 @@ def provider_catalog(user_id: UUID) -> list[dict[str, Any]]:
                 )
                 .order_by(ExternalSystemDefinition.name)
             ).scalars().all()
-            definitions = [_definition_view(row, include_config=False) for row in rows]
+            definitions_by_provider: dict[str, list[dict[str, Any]]] = {}
+            for row in rows:
+                definitions_by_provider.setdefault(row.provider, []).append(
+                    _definition_view(row, include_config=False)
+                )
     except Exception:
-        definitions = []
+        definitions_by_provider = {}
     key_ok = crypto_configured()
-    return [{
+    return [
-        "provider": "factory_mes",
+        {
-        "title": "Factory MES",
+            "provider": spec.provider,
-        "connector": "openapi",
+            "title": spec.title,
-        "configured": bool(definitions and key_ok),
+            "connector": spec.connector,
+            "default_auth_type": spec.default_auth_type,
+            "allowed_auth_types": list(spec.allowed_auth_types),
+            "configured": bool(definitions_by_provider.get(spec.provider) and key_ok),
             "reason": "" if key_ok else "ZCBOT_CREDENTIAL_MASTER_KEY 未配置或少于 32 字符",
-        "definitions": definitions,
+            "definitions": definitions_by_provider.get(spec.provider, []),
-    }]
+        }
+        for spec in provider_specs()
+    ]
 
 
 def list_external_system_definitions() -> list[dict[str, Any]]:
@@ -174,14 +188,16 @@ def create_external_system_definition(
 ) -> dict[str, Any]:
     provider = (provider or "").strip()
     name = (name or "").strip()
-    if provider != "factory_mes":
+    try:
-        raise ExternalSystemError("首版只支持 factory_mes")
+        get_provider(provider)
+    except ValueError as exc:
+        raise ExternalSystemError(str(exc)) from exc
     if not name or len(name) > 80:
         raise ExternalSystemError("系统名称不能为空且不能超过 80 字符")
     row = ExternalSystemDefinition(
         provider=provider,
         name=name,
-        config=_normalized_config(config),
+        config=_normalized_config(provider, config),
         enabled=bool(enabled),
         access_mode=_validate_access_mode(access_mode),
         created_by=admin_user_id,
@@ -222,7 +238,7 @@ def update_external_system_definition(
             if row is None:
                 raise ExternalSystemError("external system definition not found")
             row.name = name
-            row.config = _normalized_config(config)
+            row.config = _normalized_config(row.provider, config)
             row.enabled = bool(enabled)
             row.access_mode = _validate_access_mode(access_mode)
             if row.access_mode == "selected":
@@ -286,48 +302,73 @@ def get_definition_for_user(user_id: UUID, definition_id: UUID) -> ExternalSyste
 
 
 def _client(
-    provider: str, username: str, password: str, config: dict[str, Any]
+    provider: str,
-) -> FactoryMesClient:
+    credentials: dict[str, str],
-    if provider != "factory_mes":
+    config: dict[str, Any],
-        raise ExternalSystemError(f"unsupported external system provider: {provider}")
+    *,
-    return FactoryMesClient(username, password, _factory_config(config))
+    cache_namespace: str = "",
+) -> OpenApiClient:
+    spec = get_provider(provider)
+    if spec.connector != "openapi":
+        raise ExternalSystemError(f"unsupported external system connector: {spec.connector}")
+    return OpenApiClient(
+        credentials,
+        _runtime_config(provider, config),
+        cache_namespace=cache_namespace,
+    )
 
 
-def _credentials(username: str, password: str) -> dict[str, str]:
+def _credential_values(
-    username = (username or "").strip()
+    provider: str, config: dict[str, Any], credentials: dict[str, str]
-    if not username or not password:
+) -> dict[str, str]:
-        raise ExternalSystemError("用户名和密码不能为空")
+    fields = credential_fields(provider, config)
+    normalized = {
+        field["name"]: str(credentials.get(field["name"]) or "").strip()
+        for field in fields
+    }
+    missing = [field["label"] for field in fields if not normalized[field["name"]]]
+    if missing:
+        raise ExternalSystemError("请填写" + "、".join(missing))
+    return normalized
+
+
+def _credentials(
+    provider: str, config: dict[str, Any], credentials: dict[str, str]
+) -> dict[str, str]:
+    normalized = _credential_values(provider, config, credentials)
     try:
-        return {"username": encrypt_secret(username), "password": encrypt_secret(password)}
+        return {name: encrypt_secret(value) for name, value in normalized.items()}
     except (RuntimeError, ValueError) as exc:
         raise ExternalSystemError(str(exc)) from exc
 
 
-def credentials_for(row: ExternalSystem) -> tuple[str, str]:
+def credentials_for(row: ExternalSystem) -> dict[str, str]:
     try:
-        return (
+        return {name: decrypt_secret(value) for name, value in row.credentials.items()}
-            decrypt_secret(row.credentials["username"]),
+    except (AttributeError, RuntimeError) as exc:
-            decrypt_secret(row.credentials["password"]),
-        )
-    except (KeyError, RuntimeError) as exc:
         raise ExternalSystemError(str(exc)) from exc
 
 
-def client_for_external_system(row: ExternalSystem) -> FactoryMesClient:
+def client_for_external_system(row: ExternalSystem) -> OpenApiClient:
     definition = get_definition_for_user(row.user_id, row.definition_id)
-    username, password = credentials_for(row)
+    return _client(
-    return _client(definition.provider, username, password, definition.config or {})
+        definition.provider,
+        credentials_for(row),
+        definition.config or {},
+        cache_namespace=f"{definition.definition_id}:{row.user_id}",
+    )
 
 
 def _view(row: ExternalSystem, definition: ExternalSystemDefinition) -> dict[str, Any]:
     try:
-        username, _ = credentials_for(row)
+        credentials = credentials_for(row)
-        masked = mask_username(username)
+        identity = credentials.get("username") or next(iter(credentials.values()))
+        masked = mask_username(identity) if credentials.get("username") else "***"
         credential_ok = True
-    except ExternalSystemError:
+    except (ExternalSystemError, StopIteration):
         masked = "***"
         credential_ok = False
-    runtime_config = _factory_config(definition.config or {})
+    runtime_config = _runtime_config(definition.provider, definition.config or {})
     return {
         "external_system_id": str(row.external_system_id),
         "definition_id": str(row.definition_id),
@@ -338,6 +379,7 @@ def _view(row: ExternalSystem, definition: ExternalSystemDefinition) -> dict[str
         "status": row.status if definition.enabled else "disabled",
         "username_masked": masked,
         "credential_configured": credential_ok,
+        "credential_fields": credential_fields(definition.provider, definition.config or {}),
         "query_guidance": runtime_config.query_guidance,
         "recommended_operation_ids": list(runtime_config.recommended_operation_ids),
         "last_verified_at": row.last_verified_at.isoformat() if row.last_verified_at else None,
@@ -381,8 +423,9 @@ def create_external_system(
     *,
     definition_id: UUID,
     name: str,
-    username: str,
+    credentials: Optional[dict[str, str]] = None,
-    password: str,
+    username: str = "",
+    password: str = "",
 ) -> dict[str, Any]:
     if not crypto_configured():
         raise ExternalSystemError("ZCBOT_CREDENTIAL_MASTER_KEY 未配置或少于 32 字符")
@@ -390,9 +433,19 @@ def create_external_system(
     name = (name or definition.name).strip()
     if not name or len(name) > 80:
         raise ExternalSystemError("连接名称不能为空且不能超过 80 字符")
+    plain = _credential_values(
+        definition.provider,
+        definition.config or {},
+        credentials or {"username": username, "password": password},
+    )
     try:
-        probe = _client(definition.provider, username.strip(), password, definition.config).test_connection()
+        probe = _client(
-    except FactoryMesError as exc:
+            definition.provider,
+            plain,
+            definition.config,
+            cache_namespace=f"{definition.definition_id}:{user_id}",
+        ).test_connection()
+    except OpenApiError as exc:
         raise ExternalSystemError(str(exc)) from exc
     try:
         with session_scope() as s:
@@ -403,34 +456,49 @@ def create_external_system(
                 )
             ).scalar_one_or_none()
             if row is not None and row.status != "pending":
-                raise ExternalSystemError("该 MES 已连接，请使用更新凭据")
+                raise ExternalSystemError("该外部系统已连接，请使用更新凭据")
             if row is None:
                 row = ExternalSystem(
                     user_id=user_id,
                     definition_id=definition.definition_id,
                     provider=definition.provider,
-                    connector="openapi",
+                    connector=get_provider(definition.provider).connector,
                 )
                 s.add(row)
             row.name = name
-            row.credentials = _credentials(username, password)
+            row.credentials = _credentials(definition.provider, definition.config or {}, plain)
             row.config = {"operation_count": probe.get("operation_count", 0)}
             row.status = "active"
             row.last_verified_at = datetime.now(timezone.utc)
             s.flush()
             return _view(row, definition)
     except IntegrityError as exc:
-        raise ExternalSystemError("同名 MES 连接已存在") from exc
+        raise ExternalSystemError("同名外部系统连接已存在") from exc
 
 
 def update_external_system_credentials(
-    user_id: UUID, system_id: UUID, *, username: str, password: str
+    user_id: UUID,
+    system_id: UUID,
+    *,
+    credentials: Optional[dict[str, str]] = None,
+    username: str = "",
+    password: str = "",
 ) -> dict[str, Any]:
     row = get_external_system(user_id, system_id)
     definition = get_definition(row.definition_id, enabled_only=True)
+    plain = _credential_values(
+        definition.provider,
+        definition.config or {},
+        credentials or {"username": username, "password": password},
+    )
     try:
-        probe = _client(definition.provider, username.strip(), password, definition.config).test_connection()
+        probe = _client(
-    except FactoryMesError as exc:
+            definition.provider,
+            plain,
+            definition.config,
+            cache_namespace=f"{definition.definition_id}:{user_id}",
+        ).test_connection()
+    except OpenApiError as exc:
         raise ExternalSystemError(str(exc)) from exc
     with session_scope() as s:
         current = s.execute(
@@ -439,7 +507,7 @@ def update_external_system_credentials(
                 ExternalSystem.user_id == user_id,
             )
         ).scalar_one()
-        current.credentials = _credentials(username, password)
+        current.credentials = _credentials(definition.provider, definition.config or {}, plain)
         current.config = {**(current.config or {}), "operation_count": probe.get("operation_count", 0)}
         current.status = "active"
         current.last_verified_at = datetime.now(timezone.utc)
@@ -453,7 +521,7 @@ def test_external_system(user_id: UUID, system_id: UUID) -> dict[str, Any]:
     try:
         probe = client_for_external_system(row).test_connection()
         ok = True
-    except (ExternalSystemError, FactoryMesError) as exc:
+    except (ExternalSystemError, OpenApiError) as exc:
         error = str(exc)
     with session_scope() as s:
         current = s.execute(

tests/test_external_systems.py

@@ -164,6 +164,46 @@ class FactoryOpenApiConnectorTests(unittest.TestCase):
                 "openapi_url": "https://factory.invalid/swagger.json",
             })
 
+    def test_generic_api_key_auth_uses_declared_header_without_login(self):
+        from core.external_systems.openapi import OpenApiClient, OpenApiConfig, _SPEC_CACHE
+        from core.external_systems.registry import credential_fields, merged_config
+
+        _SPEC_CACHE.clear()
+        config = merged_config("generic_openapi", {
+            "base_url": "https://erp.invalid",
+            "openapi_url": "https://erp.invalid/openapi.json",
+            "auth_type": "api_key",
+            "auth_header_name": "X-ERP-Key",
+            "auth_header_template": "Key {token}",
+        })
+        client = OpenApiClient({"api_key": "private-key"}, OpenApiConfig.from_mapping(config))
+        http = _Http()
+        with patch.object(client, "_client", return_value=http):
+            result = client.test_connection()
+        self.assertGreater(result["operation_count"], 0)
+        self.assertFalse(any(call[0] == "POST" for call in http.calls))
+        get_call = next(call for call in http.calls if call[0] == "GET")
+        self.assertEqual(get_call[2]["headers"], {"X-ERP-Key": "Key private-key"})
+        self.assertEqual(credential_fields("generic_openapi", config)[0]["name"], "api_key")
+
+    def test_openapi_spec_cache_is_isolated_by_connection_namespace(self):
+        from core.external_systems.openapi import OpenApiClient, OpenApiConfig, _SPEC_CACHE
+        from core.external_systems.registry import merged_config
+
+        _SPEC_CACHE.clear()
+        config = OpenApiConfig.from_mapping(merged_config("generic_openapi", {
+            "base_url": "https://erp.invalid",
+            "openapi_url": "https://erp.invalid/openapi.json",
+            "auth_type": "bearer_token",
+        }))
+        for namespace in ("definition:user-a", "definition:user-b"):
+            client = OpenApiClient({"token": namespace}, config, cache_namespace=namespace)
+            http = _Http()
+            with patch.object(client, "_client", return_value=http):
+                client.test_connection()
+            self.assertTrue(any(call[0] == "GET" for call in http.calls))
+        self.assertEqual(len(_SPEC_CACHE), 2)
+
     def test_search_discovers_operation_without_exposing_credentials(self):
         from core.external_systems.factory import FactoryMesClient
 

tests/test_web_routes_nodb.py

@@ -167,6 +167,22 @@ class ExternalSystemRoutesTests(unittest.TestCase):
         self.assertEqual(create.call_args.args[0], _UID)
         self.assertEqual(create.call_args.kwargs["definition_id"], definition_id)
 
+    def test_create_accepts_dynamic_credentials(self):
+        created = {"external_system_id": str(uuid.uuid4()), "username_masked": "***"}
+        definition_id = uuid.uuid4()
+        with patch("web.routers.external_systems.create_external_system", return_value=created) as create:
+            response = _client.post(
+                "/v1/external-systems",
+                headers=_AUTH,
+                json={
+                    "definition_id": str(definition_id),
+                    "name": "LIMS",
+                    "credentials": {"api_key": "secret-key"},
+                },
+            )
+        self.assertEqual(response.status_code, 201)
+        self.assertEqual(create.call_args.kwargs["credentials"], {"api_key": "secret-key"})
+
     def test_invalid_connection_id_is_not_forwarded_to_service(self):
         with patch("web.routers.external_systems.test_external_system") as test_connection:
             r = _client.post("/v1/external-systems/not-a-uuid/test", headers=_AUTH)

web/admin.py

@@ -188,6 +188,12 @@ class ExternalSystemDefinitionRequest(BaseModel):
     base_url: str
     openapi_url: str
     login_path: str = "/api/auth/token/"
+    auth_type: str = "password_jwt"
+    username_field: str = "username"
+    password_field: str = "password"
+    token_field: str = "access"
+    auth_header_name: str = "Authorization"
+    auth_header_template: str = "Bearer {token}"
     allowed_post_operations: list[str] = Field(default_factory=list)
     timeout_seconds: float = 15
     max_result_bytes: int = 65536
@@ -208,6 +214,12 @@ def _external_definition_config(body: ExternalSystemDefinitionRequest) -> dict[s
         "base_url": body.base_url,
         "openapi_url": body.openapi_url,
         "login_path": body.login_path,
+        "auth_type": body.auth_type,
+        "username_field": body.username_field,
+        "password_field": body.password_field,
+        "token_field": body.token_field,
+        "auth_header_name": body.auth_header_name,
+        "auth_header_template": body.auth_header_template,
         "allowed_post_operations": body.allowed_post_operations,
         "timeout_seconds": body.timeout_seconds,
         "max_result_bytes": body.max_result_bytes,

web/routers/external_systems.py

@@ -54,6 +54,7 @@ def register_external_system_routes(app, *, require_user) -> None:
                 user_id,
                 definition_id=body.definition_id,
                 name=body.name,
+                credentials=body.credentials or None,
                 username=body.username,
                 password=body.password,
             )
@@ -70,6 +71,7 @@ def register_external_system_routes(app, *, require_user) -> None:
             return update_external_system_credentials(
                 user_id,
                 _uuid(system_id),
+                credentials=body.credentials or None,
                 username=body.username,
                 password=body.password,
             )

web/schemas.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 from typing import Optional
 from uuid import UUID
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 
 class TaskCreateRequest(BaseModel):
@@ -97,10 +97,12 @@ class KbCreateRequest(BaseModel):
 class ExternalSystemCreateRequest(BaseModel):
     definition_id: UUID
     name: str = ""
-    username: str
+    credentials: dict[str, str] = Field(default_factory=dict)
-    password: str
+    username: str = ""  # deprecated: 兼容旧版 Factory 客户端
+    password: str = ""  # deprecated: 兼容旧版 Factory 客户端
 
 
 class ExternalSystemCredentialsRequest(BaseModel):
-    username: str
+    credentials: dict[str, str] = Field(default_factory=dict)
-    password: str
+    username: str = ""  # deprecated
+    password: str = ""  # deprecated

web/static/dev.html

@@ -1685,12 +1685,11 @@
       <div id="ext-provider" class="muted" style="font-size:12px;margin-bottom:10px;">加载中…</div>
       <div id="ext-list"><div class="muted">加载中…</div></div>
       <form id="ext-form">
-        <div style="font-weight:600;margin-bottom:10px;" id="ext-form-title">连接 Factory MES</div>
+        <div style="font-weight:600;margin-bottom:10px;" id="ext-form-title">连接外部系统</div>
         <div id="ext-form-grid">
-          <label class="wide">MES 系统<select id="ext-definition"></select></label>
+          <label class="wide">外部系统<select id="ext-definition"></select></label>
-          <label class="wide">连接名称<input id="ext-name" value="Factory MES" maxlength="80" autocomplete="off"></label>
+          <label class="wide">连接名称<input id="ext-name" maxlength="80" autocomplete="off"></label>
-          <label>MES 用户名<input id="ext-username" autocomplete="username"></label>
+          <div id="ext-credential-fields" class="wide" style="display:grid;grid-template-columns:1fr 1fr;gap:8px;"></div>
-          <label>MES 密码<input id="ext-password" type="password" autocomplete="current-password"></label>
         </div>
         <div id="ext-err" class="err" style="margin-top:8px;"></div>
         <div id="ext-form-actions">

web/static/js/admin.js

@@ -167,7 +167,7 @@ function renderExternalDefinitions() {
   const rows = externalDefinitions.map(r => {
     const cfg = r.config || {};
     return `<tr data-definition-id="${escapeHtml(r.definition_id)}">`
-      + `<td>${escapeHtml(r.name)}${r.enabled ? "" : ' <span class="chip">停用</span>'}`
+      + `<td>${escapeHtml(r.name)} <span class="chip">${escapeHtml(r.provider_title || r.provider)}</span>${r.enabled ? "" : ' <span class="chip">停用</span>'}`
       + ` <span class="chip">${r.access_mode === "all" ? "全部用户" : `指定 ${((r.selected_user_ids || []).length)} 人`}</span></td>`
       + `<td class="email" title="${escapeHtml(cfg.base_url || "")}">${escapeHtml(r.host || cfg.base_url || "—")}</td>`
       + `<td class="num">${(cfg.allowed_post_operations || []).length}</td>`
@@ -175,12 +175,19 @@ function renderExternalDefinitions() {
       + `</tr>`;
   }).join("") || `<tr><td colspan="4" class="empty">尚未配置外部系统</td></tr>`;
   $("s-external").innerHTML = `<div class="card"><div class="card-head"><h2>外部系统目录</h2>`
-    + `<span class="sublabel">公共地址由管理员维护；用户只提交自己的账号密码</span></div>`
+    + `<span class="sublabel">标准 OpenAPI 系统可直接配置；用户只提交该系统要求的凭据</span></div>`
     + `<form id="ext-admin-form" style="display:grid;grid-template-columns:1fr 1fr;gap:8px;margin-bottom:14px;">`
+    + `<label>系统类型<select id="exa-provider"><option value="factory_mes">Factory MES</option><option value="generic_openapi">通用 OpenAPI 系统</option></select></label>`
+    + `<label>认证方式<select id="exa-auth"><option value="password_jwt">用户名密码换取 Token</option><option value="api_key">API Key</option><option value="bearer_token">Bearer Token</option></select></label>`
     + `<label>系统名称<input id="exa-name" required placeholder="Factory MES"></label>`
     + `<label>Base URL<input id="exa-base" required placeholder="https://factory.example.com"></label>`
     + `<label>Swagger / OpenAPI URL<input id="exa-spec" required placeholder="https://factory.example.com/swagger.json"></label>`
     + `<label>登录路径<input id="exa-login" value="/api/auth/token/"></label>`
+    + `<label>Token 字段路径<input id="exa-token-field" value="access" placeholder="data.access_token"></label>`
+    + `<label>用户名字段<input id="exa-username-field" value="username"></label>`
+    + `<label>密码字段<input id="exa-password-field" value="password"></label>`
+    + `<label>认证 Header<input id="exa-auth-header" value="Authorization"></label>`
+    + `<label>Header 模板<input id="exa-auth-template" value="Bearer {token}"></label>`
     + `<label style="grid-column:1/-1;">允许的只读 POST operationId（逗号分隔）<input id="exa-post" placeholder="bi_dataset_exec"></label>`
     + `<label style="grid-column:1/-1;">推荐查询入口 operationId（逗号分隔）<input id="exa-recommended" value="bi_dataset_list, bi_dataset_exec"></label>`
     + `<label style="grid-column:1/-1;">查询规划提示`
@@ -203,10 +210,13 @@ function renderExternalDefinitions() {
   $("exa-guidance").value = DEFAULT_EXTERNAL_QUERY_GUIDANCE;
   updateExternalGuidanceSummary();
   $("exa-guidance-edit").onclick = editExternalGuidance;
+  $("exa-provider").onchange = applyExternalProviderDefaults;
+  $("exa-auth").onchange = applyExternalAuthDefaults;
   $("exa-access").onchange = () => {
     $("exa-users-wrap").hidden = $("exa-access").value !== "selected";
   };
   $("exa-cancel").onclick = () => { externalEditingId = ""; renderExternalDefinitions(); };
+  updateExternalAuthForm();
   $("s-external").onclick = (e) => {
     const tr = e.target.closest("tr[data-definition-id]");
     if (!tr) return;
@@ -217,6 +227,33 @@ function renderExternalDefinitions() {
   };
 }
 
+function updateExternalAuthForm() {
+  const provider = $("exa-provider").value;
+  if (provider === "factory_mes") $("exa-auth").value = "password_jwt";
+  $("exa-auth").disabled = provider === "factory_mes";
+  const login = $("exa-auth").value === "password_jwt";
+  for (const id of ["exa-login", "exa-token-field", "exa-username-field", "exa-password-field"]) {
+    $(id).closest("label").hidden = !login;
+  }
+}
+
+function applyExternalProviderDefaults() {
+  const factory = $("exa-provider").value === "factory_mes";
+  $("exa-auth").value = "password_jwt";
+  $("exa-recommended").value = factory ? "bi_dataset_list, bi_dataset_exec" : "";
+  $("exa-guidance").value = factory ? DEFAULT_EXTERNAL_QUERY_GUIDANCE : "";
+  $("exa-name").placeholder = factory ? "Factory MES" : "ERP / LIMS / 其他系统";
+  applyExternalAuthDefaults();
+  updateExternalGuidanceSummary();
+}
+
+function applyExternalAuthDefaults() {
+  const auth = $("exa-auth").value;
+  $("exa-auth-header").value = auth === "api_key" ? "X-API-Key" : "Authorization";
+  $("exa-auth-template").value = auth === "api_key" ? "{token}" : "Bearer {token}";
+  updateExternalAuthForm();
+}
+
 function updateExternalGuidanceSummary() {
   const text = ($("exa-guidance").value || "").trim();
   $("exa-guidance-summary").textContent = text
@@ -242,10 +279,19 @@ async function editExternalGuidance() {
 function fillExternalDefinition(row) {
   externalEditingId = row.definition_id;
   const cfg = row.config || {};
+  $("exa-provider").value = row.provider || "factory_mes";
+  $("exa-provider").disabled = true;
+  $("exa-auth").value = cfg.auth_type || "password_jwt";
   $("exa-name").value = row.name || "";
   $("exa-base").value = cfg.base_url || "";
   $("exa-spec").value = cfg.openapi_url || "";
   $("exa-login").value = cfg.login_path || "/api/auth/token/";
+  $("exa-token-field").value = cfg.token_field || "access";
+  $("exa-username-field").value = cfg.username_field || "username";
+  $("exa-password-field").value = cfg.password_field || "password";
+  $("exa-auth-header").value = cfg.auth_header_name || "Authorization";
+  $("exa-auth-template").value = cfg.auth_header_template || "Bearer {token}";
+  updateExternalAuthForm();
   $("exa-post").value = (cfg.allowed_post_operations || []).join(", ");
   $("exa-recommended").value = (
     cfg.recommended_operation_ids || ["bi_dataset_list", "bi_dataset_exec"]
@@ -265,11 +311,17 @@ function fillExternalDefinition(row) {
 async function saveExternalDefinition(e) {
   e.preventDefault();
   const body = {
-    provider: "factory_mes",
+    provider: $("exa-provider").value,
     name: $("exa-name").value.trim(),
     base_url: $("exa-base").value.trim(),
     openapi_url: $("exa-spec").value.trim(),
     login_path: $("exa-login").value.trim() || "/api/auth/token/",
+    auth_type: $("exa-auth").value,
+    username_field: $("exa-username-field").value.trim() || "username",
+    password_field: $("exa-password-field").value.trim() || "password",
+    token_field: $("exa-token-field").value.trim() || "access",
+    auth_header_name: $("exa-auth-header").value.trim() || "Authorization",
+    auth_header_template: $("exa-auth-template").value || "Bearer {token}",
     allowed_post_operations: $("exa-post").value.split(",").map(x => x.trim()).filter(Boolean),
     recommended_operation_ids: $("exa-recommended").value.split(",").map(x => x.trim()).filter(Boolean),
     query_guidance: $("exa-guidance").value.trim(),

web/static/js/external_systems.js

@@ -1,4 +1,4 @@
-// 外部系统连接管理：凭据只提交、不回显；查询能力在下一轮 build_agent 时按用户动态挂载。
+// 外部系统连接管理：凭据字段由系统定义声明，只提交、不回显。
 import { $ } from "./dom.js";
 import { api } from "./api.js";
 import { escapeHtml, fmtTime } from "./format.js";
@@ -7,18 +7,32 @@ import { dialogConfirm, message } from "./dialog.js";
 let editingId = "";
 let providerReady = false;
 let definitions = [];
+let systemsById = new Map();
+
+function selectedDefinition() {
+  return definitions.find(x => x.definition_id === $("ext-definition").value);
+}
+
+function renderCredentialFields(fields = []) {
+  $("ext-credential-fields").innerHTML = fields.map(field => `
+    <label>${escapeHtml(field.label)}<input
+      data-credential-name="${escapeHtml(field.name)}"
+      type="${field.secret ? "password" : "text"}"
+      autocomplete="${escapeHtml(field.autocomplete || "off")}" required></label>
+  `).join("");
+}
 
 function resetForm() {
   editingId = "";
-  $("ext-form-title").textContent = "连接 Factory MES";
+  $("ext-form-title").textContent = "连接外部系统";
-  $("ext-name").value = "Factory MES";
   $("ext-name").disabled = false;
   $("ext-definition").disabled = false;
-  $("ext-username").value = "";
-  $("ext-password").value = "";
   $("ext-err").textContent = "";
   $("ext-save").textContent = "连接并验证";
   $("ext-form-cancel").hidden = true;
+  const selected = selectedDefinition() || definitions[0];
+  $("ext-name").value = selected ? selected.name : "";
+  renderCredentialFields((selected || {}).credential_fields || []);
 }
 
 export function closeExternalSystemsModal() {
@@ -38,7 +52,7 @@ function cardHtml(item) {
   const checked = item.last_verified_at ? fmtTime(item.last_verified_at) : "尚未验证";
   return `<div class="ext-card" data-id="${escapeHtml(item.external_system_id)}">
     <div class="ext-card-head"><span class="ext-card-name">${escapeHtml(item.name)}</span><span class="sk-badge">${badge}</span></div>
-    <div class="meta">${escapeHtml(item.system_name || "Factory MES")} · ${escapeHtml(item.username_masked || "***")} · 最近验证 ${escapeHtml(checked)}</div>
+    <div class="meta">${escapeHtml(item.system_name || "外部系统")} · ${escapeHtml(item.username_masked || "***")} · 最近验证 ${escapeHtml(checked)}</div>
     <div class="ext-actions">
       <button type="button" class="small" data-ext-test>测试连接</button>
       <button type="button" class="small" data-ext-edit>更新凭据</button>
@@ -53,116 +67,119 @@ async function loadExternalSystems() {
   provider.textContent = "加载中…";
   list.innerHTML = '<div class="muted">加载中…</div>';
   try {
-    const [p, systems] = await Promise.all([
+    const [catalog, response] = await Promise.all([
       api("GET", "/v1/external-system-providers"),
       api("GET", "/v1/external-systems"),
     ]);
-    const factory = (p.providers || []).find((x) => x.provider === "factory_mes");
+    const providers = catalog.providers || [];
-    definitions = (factory && factory.definitions) || [];
+    definitions = providers.flatMap(item => item.definitions || []);
-    providerReady = !!(factory && factory.configured && definitions.length);
+    providerReady = definitions.length > 0 && providers.some(item => item.configured);
-    $("ext-definition").innerHTML = definitions.map(x =>
+    $("ext-definition").innerHTML = definitions.map(item =>
-      `<option value="${escapeHtml(x.definition_id)}">${escapeHtml(x.name)}${x.host ? " · " + escapeHtml(x.host) : ""}</option>`
+      `<option value="${escapeHtml(item.definition_id)}">${escapeHtml(item.name)}${item.host ? " · " + escapeHtml(item.host) : ""}</option>`
     ).join("");
-    if (!editingId && definitions.length) $("ext-name").value = definitions[0].name;
+    const rows = response.results || [];
+    systemsById = new Map(rows.map(item => [item.external_system_id, item]));
     provider.textContent = providerReady
-      ? `管理员已配置 ${definitions.length} 个 MES 系统，请选择后使用自己的 MES 账号连接`
+      ? `管理员已配置 ${definitions.length} 个外部系统，请选择后填写自己的访问凭据`
-      : `暂没有可连接的 MES 系统${factory && factory.reason ? "：" + factory.reason : "，请联系管理员配置"}`;
+      : `暂没有可连接的外部系统${providers.find(item => item.reason)?.reason ? "：" + providers.find(item => item.reason).reason : "，请联系管理员配置"}`;
     $("ext-save").disabled = !providerReady;
-    const rows = systems.results || [];
     list.innerHTML = rows.length
       ? rows.map(cardHtml).join("")
-      : '<div class="sk-empty">还没有外部系统连接。填写下方 MES 账号后，助手即可按你的 MES 权限查询。</div>';
+      : '<div class="sk-empty">还没有外部系统连接。填写下方凭据后，助手即可按你的权限查询。</div>';
-  } catch (e) {
+    resetForm();
+  } catch (error) {
     providerReady = false;
     provider.textContent = "加载失败";
-    list.innerHTML = `<div class="err">${escapeHtml(e.message)}</div>`;
+    list.innerHTML = `<div class="err">${escapeHtml(error.message)}</div>`;
     $("ext-save").disabled = true;
   }
 }
 
 $("hd-external").onclick = openExternalSystemsModal;
 $("ext-close").onclick = closeExternalSystemsModal;
-$("external-modal").addEventListener("click", (e) => {
+$("external-modal").addEventListener("click", event => {
-  if (e.target.id === "external-modal") closeExternalSystemsModal();
+  if (event.target.id === "external-modal") closeExternalSystemsModal();
 });
 $("ext-form-cancel").onclick = resetForm;
 $("ext-definition").onchange = () => {
   if (editingId) return;
-  const selected = definitions.find(x => x.definition_id === $("ext-definition").value);
+  const selected = selectedDefinition();
   if (selected) $("ext-name").value = selected.name;
+  renderCredentialFields((selected || {}).credential_fields || []);
 };
 
-$("ext-form").addEventListener("submit", async (e) => {
+$("ext-form").addEventListener("submit", async event => {
-  e.preventDefault();
+  event.preventDefault();
   if (!providerReady) return;
-  const username = $("ext-username").value.trim();
-  const password = $("ext-password").value;
   const name = $("ext-name").value.trim();
   const definitionId = $("ext-definition").value;
-  if (!username || !password || (!editingId && (!name || !definitionId))) {
+  const credentials = Object.fromEntries(
-    $("ext-err").textContent = "请填写连接名称、MES 用户名和密码";
+    Array.from($("ext-credential-fields").querySelectorAll("[data-credential-name]"))
+      .map(input => [input.dataset.credentialName, input.value]),
+  );
+  if (Object.values(credentials).some(value => !value.trim()) || (!editingId && (!name || !definitionId))) {
+    $("ext-err").textContent = "请填写连接名称和全部访问凭据";
     return;
   }
-  const btn = $("ext-save");
+  const button = $("ext-save");
-  btn.disabled = true;
+  button.disabled = true;
-  btn.textContent = "正在验证…";
+  button.textContent = "正在验证…";
   $("ext-err").textContent = "";
   try {
     if (editingId) {
-      await api("PUT", `/v1/external-systems/${editingId}/credentials`, { username, password });
+      await api("PUT", `/v1/external-systems/${editingId}/credentials`, { credentials });
-      message("Factory MES 凭据已更新", "success");
+      message("外部系统凭据已更新", "success");
     } else {
-      await api("POST", "/v1/external-systems", {
+      await api("POST", "/v1/external-systems", { definition_id: definitionId, name, credentials });
-        definition_id: definitionId, name, username, password,
+      message("外部系统已连接；下一轮对话即可使用", "success");
-      });
-      message("Factory MES 已连接；下一轮对话即可使用", "success");
     }
     resetForm();
     await loadExternalSystems();
-  } catch (err) {
+  } catch (error) {
-    $("ext-err").textContent = err.message;
+    $("ext-err").textContent = error.message;
   } finally {
-    btn.disabled = !providerReady;
+    button.disabled = !providerReady;
-    btn.textContent = editingId ? "验证并更新" : "连接并验证";
+    button.textContent = editingId ? "验证并更新" : "连接并验证";
   }
 });
 
-$("ext-list").addEventListener("click", async (e) => {
+$("ext-list").addEventListener("click", async event => {
-  const card = e.target.closest(".ext-card");
+  const card = event.target.closest(".ext-card");
   if (!card) return;
   const id = card.dataset.id;
-  if (e.target.closest("[data-ext-edit]")) {
+  if (event.target.closest("[data-ext-edit]")) {
+    const system = systemsById.get(id);
     editingId = id;
-    $("ext-form-title").textContent = "更新 Factory MES 凭据";
+    $("ext-form-title").textContent = "更新外部系统凭据";
-    $("ext-name").value = card.querySelector(".ext-card-name").textContent;
+    $("ext-name").value = system.name;
     $("ext-name").disabled = true;
+    $("ext-definition").value = system.definition_id;
     $("ext-definition").disabled = true;
-    $("ext-username").value = "";
+    renderCredentialFields(system.credential_fields || []);
-    $("ext-password").value = "";
     $("ext-save").textContent = "验证并更新";
     $("ext-form-cancel").hidden = false;
-    $("ext-username").focus();
+    $("ext-credential-fields").querySelector("input")?.focus();
     return;
   }
-  if (e.target.closest("[data-ext-test]")) {
+  if (event.target.closest("[data-ext-test]")) {
-    const btn = e.target.closest("button");
+    const button = event.target.closest("button");
-    btn.disabled = true;
+    button.disabled = true;
     try {
       const result = await api("POST", `/v1/external-systems/${id}/test`);
       message(result.ok ? `连接正常，可用接口 ${result.operation_count || 0} 个` : `连接失败：${result.error}`, result.ok ? "success" : "error");
       await loadExternalSystems();
-    } catch (err) { message("测试失败：" + err.message, "error"); }
+    } catch (error) { message("测试失败：" + error.message, "error"); }
-    finally { btn.disabled = false; }
+    finally { button.disabled = false; }
     return;
   }
-  if (e.target.closest("[data-ext-delete]")) {
+  if (event.target.closest("[data-ext-delete]")) {
     const name = card.querySelector(".ext-card-name").textContent;
-    if (!await dialogConfirm({ title:"断开外部系统", message:`断开「${name}」？保存的 MES 密文凭据和连接配置将被清除。`, okText:"断开", danger:true })) return;
+    if (!await dialogConfirm({ title:"断开外部系统", message:`断开「${name}」？保存的密文凭据和连接配置将被清除。`, okText:"断开", danger:true })) return;
     try {
       await api("DELETE", `/v1/external-systems/${id}`);
       message("外部系统已断开", "success");
       resetForm();
       await loadExternalSystems();
-    } catch (err) { message("断开失败：" + err.message, "error"); }
+    } catch (error) { message("断开失败：" + error.message, "error"); }
   }
 });