feat(mcp): 支持用户个人凭证并优化配置交互

ARCHITECTURE.md

@@ -16,7 +16,7 @@ Yuxi 是一个面向 RAG、知识图谱和多智能体工作流的知识库平
 - `api`：FastAPI API 服务，挂载 `backend/yuxi` 和测试目录；API 与 worker 的源码修改通过显式重启生效。
 - `worker`：ARQ worker，执行已经派发的 AgentRun 与注册的 后台作业，并周期触发用户自建 Agent 定时任务；三者分别使用 PostgreSQL 中的运行租约、任务租约和调度锁闭合并发与恢复。
 - `storage-init`：无网络的一次性进程，准备 API/worker/app-init 与 Milvus 挂载根的属主和模式，不递归修改已有文件；消费进程等待其成功退出。
-- `app-init`：普通用户运行的部署任务，在 PostgreSQL advisory lock 内先建立或校验当前 Schema，再同步系统配置定义、内置供应商、Agent、MCP 与 Skill 来源文件。API 与 worker 等待全部成功。空库基线为 business 1、knowledge 2，直接包含个人 MCP 授权表、128 位用户归属字段及当前约束。当前版本跳过结构初始化，其他版本明确拒绝；历史部署的结构调整由部署侧显式完成；知识域 v1 路径目录通过离线审计/转换入口升级 v2，运行进程不自动迁移。共享同步失败保留已提交内容，修正后幂等重跑。
+- `app-init`：普通用户运行的部署任务，在 PostgreSQL advisory lock 内先建立或校验当前 Schema，再同步系统配置定义、内置供应商、Agent、MCP 与 Skill 来源文件。API 与 worker 等待全部成功。空库基线为 business 2、knowledge 2，直接包含个人 MCP 授权表、128 位用户归属字段及当前约束。当前版本跳过结构初始化，其他版本明确拒绝；历史部署的结构调整由部署侧显式完成；知识域 v1 路径目录通过离线审计/转换入口升级 v2，运行进程不自动迁移。共享同步失败保留已提交内容，修正后幂等重跑。
 - Cube Linux/KVM 执行节点：每个用户拥有一个持久普通账号 VM，Workspace 与授权 Skills 共享宿主文件。
 - `sandbox-runner`：选择 container 后端时由该独立进程持有 Docker socket，为每个 Run 懒创建临时容器并执行命令；API/worker 只调用认证 HTTP。挂载、镜像和资源限制由 runner 拥有，Run 归属与清理责任仍由 PostgreSQL 拥有。
 - `postgres`：业务数据、知识库元数据、持久 Input 队列、Turn/Run 与 LangGraph checkpoint。
@@ -51,7 +51,7 @@ Yuxi 默认交付完整知识能力。部署开关 `YUXI_KNOWLEDGE_ENABLED` 由
 - `modules/identity` 拥有用户、部门、权限、凭据与 OIDC 账号用例，以及 Public API 的 Key 校验与 App/end_user 身份解析；`modules/extensions` 拥有 Skills、MCP 与工具目录；`modules/models` 拥有模型适配和供应商配置。
 - `modules/schedules` 拥有用户定时 Agent 定义和 occurrence；`modules/background_jobs` 拥有任务登记、执行方上报的观察记录和取消意图；后台作业模块的 dispatch 与 registry 拥有提交和定义，`workers` 拥有领取、执行和失联维护；`modules/system` 拥有系统配置、Dashboard 与跨域共享资源所有权治理。
 
-`modules/extensions/mcp` 拥有远程服务器、个人 OAuth 连接和加密凭据。超管固定授权服务器信任根，浏览器事务绑定用户、浏览器和配置修订；回调结果由当前 JWT 用户确认后发布。PostgreSQL 连接行锁串行化刷新与断开，世代阻止旧回调恢复授权。MCP service 根据可信 runtime uid 统一解析本人的 MCP 可用性；Agent 与 Skill 只消费可用服务和工具，未连接时保持普通聊天可用，相关 Skill 依赖在执行边界关闭。工具目录归本次 Run 的 Context，服务按需发现一次后在运行内复用，新 Run 重新发现；每次出站复查当前用户、配置、工具和连接。 配置、协议、授权事务与个人 HTTP 适配分属对应模块，具体链路和适配边界见 [MCP 授权与运行身份](docs/mechanisms/mcp.md)。
+`modules/extensions/mcp` 拥有远程服务器、个人 OAuth/静态凭证连接和加密存储。超管固定授权服务器信任根，浏览器事务绑定用户、浏览器和配置修订；回调结果由当前 JWT 用户确认后发布。PostgreSQL 连接行锁串行化刷新与断开，世代阻止旧回调恢复授权。MCP service 根据可信 runtime uid 统一解析本人的 MCP 可用性；Agent 与 Skill 只消费可用服务和工具，未连接时保持普通聊天可用，相关 Skill 依赖在执行边界关闭。工具目录归本次 Run 的 Context，服务按需发现一次后在运行内复用，新 Run 重新发现；每次出站复查当前用户、配置、工具和连接。 配置、协议、授权事务与个人 HTTP 适配分属对应模块，具体链路和适配边界见 [MCP 授权与运行身份](docs/mechanisms/mcp.md)。
 
 `modules/extensions/web` 拥有网络搜索、网页抓取与供应商适配；工具和管理员测试复用同一服务，每次调用读取当前有效配置，未配置的能力不进入运行工具集合。凭据与默认供应商复用 system 的持久设置和脱敏边界。
 
@@ -84,7 +84,7 @@ Yuxi 默认交付完整知识能力。部署开关 `YUXI_KNOWLEDGE_ENABLED` 由
 - `apis` 是集中 HTTP 边界，复用 `base.js` 的请求、鉴权和错误处理。模块不能依赖 app/pages，shared 不能依赖业务模块或 API；ESLint 检查别名、相对路径和动态导入。跨模块只使用 `modules/<domain>` 的模型/用例公开入口或 `modules/<domain>/ui/**/*.vue` 公开组件，不得直达 `model/` 或其他内部路径。
 - 新 TypeScript 逻辑经 strict 类型检查，build 先执行 typecheck；现有 JavaScript 逐步迁移。sessionRuntime 按 Thread 拥有一份 item 状态，历史分页、SSE 与恢复快照写入同一 reducer，展示按需派生。sessionInput 拥有冻结提交、幂等重试和接收交接；输入队列观察者确认 Input 消费并触发持久运行恢复，共享 runtime 根据当前 Turn 选择唯一运行订阅。SessionWorkspace 负责编辑器草稿、附件消费与面板展示。视图存活期间的草稿分别保留，首次进入 Thread 从持久草稿初始化。独立 Thread 阅读与后续类型迁移见[重构提案](docs/develop-guides/decisions/proposed/2026-09-30-agent-view-frontend-refactor.md)。
 
-`/` 是公开首页；登录后的核心工作区是 `/agent`。`/extensions` 对所有登录用户开放，其中 Skills 与 MCP 个人连接对普通用户可见，知识库和工具仅管理员可见，MCP 配置仅超级管理员可修改；Dashboard 仅超级管理员可访问。后端权限检查始终是最终边界，前端守卫只负责页面体验。
+`/` 是公开首页；登录后的核心工作区是 `/agent`。`/extensions` 对所有登录用户开放，其中 Skills 与 MCP 个人连接对普通用户可见，知识库和工具仅管理员可见，MCP 配置仅超级管理员可修改，Token/API Key/OAuth 凭证由用户分别配置；Dashboard 仅超级管理员可访问。后端权限检查始终是最终边界，前端守卫只负责页面体验。
 
 ## 智能体运行链路
 

backend/test/e2e/test_mcp_personal_credentials.py

@@ -0,0 +1,173 @@
+"""个人静态凭证经过实际 API、worker 和工具协议的确定性验收。"""
+
+import json
+from http.server import BaseHTTPRequestHandler
+from uuid import uuid4
+
+import asyncpg
+import pytest
+
+from test.e2e.e2e_helpers import (
+    archive_public_thread,
+    output_text,
+    postgres_dsn,
+    public_message,
+    register_replay_provider,
+    wait_for_consumed_input,
+    wait_for_terminal_turn,
+)
+from test.support.openai_replay_server import send_json_response, send_openai_sse_response, serve_replay_handler
+from test.support.public_api import delete_agent
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.e2e, pytest.mark.slow, pytest.mark.self_managed_resources]
+
+
+@pytest.mark.parametrize("mode", ["bearer", "api_key"])
+async def test_personal_credentials_reach_worker_with_own_saved_secret(e2e_client, e2e_headers, mode):
+    """工具独立核对秘密，模型只转述真实返回，最终结果绑定 result Run。"""
+    client, headers = e2e_client, e2e_headers
+    suffix = uuid4().hex[:10]
+    slug, agent, provider = ("pytest-personal-" + suffix, "pytest-personal-agent-" + suffix, "personal-replay-" + suffix)
+    secret, identity = "fixture-personal-secret-" + suffix, "fixture-personal-identity-" + suffix
+    observed, thread, turn_id = [], None, None
+    db = await asyncpg.connect(postgres_dsn())
+
+    class Handler(BaseHTTPRequestHandler):
+        """远端凭证 oracle 与模型 wire oracle 共享一次性端口。"""
+
+        def do_POST(self):
+            """MCP 核对认证头，模型核对实际工具结果且不能看到秘密。"""
+            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+            if self.path.endswith("/mcp"):
+                header = "Authorization" if mode == "bearer" else "X-API-Key"
+                expected = "Bearer " + secret if mode == "bearer" else secret
+                if self.headers.get(header) != expected:
+                    send_json_response(self, 401, {})
+                    return
+                if "id" not in body:
+                    self.send_response(202)
+                    self.send_header("Content-Length", "0")
+                    self.end_headers()
+                    return
+                method = body["method"]
+                if method == "initialize":
+                    result = {
+                        "protocolVersion": body["params"]["protocolVersion"],
+                        "capabilities": {"tools": {}},
+                        "serverInfo": {"name": "personal-fixture", "version": "1"},
+                    }
+                elif method == "tools/list":
+                    result = {
+                        "tools": [
+                            {
+                                "name": "personal_identity",
+                                "description": "读取本人身份",
+                                "inputSchema": {"type": "object", "properties": {}},
+                            }
+                        ]
+                    }
+                else:
+                    assert method == "tools/call" and body["params"]["name"] == "personal_identity"
+                    observed.append(identity)
+                    result = {"content": [{"type": "text", "text": identity}], "isError": False}
+                send_json_response(self, 200, {"jsonrpc": "2.0", "id": body["id"], "result": result})
+                return
+            assert secret not in json.dumps(body)
+            tools = [message for message in body["messages"] if message["role"] == "tool"]
+            if tools:
+                assert identity in str(tools[-1]["content"])
+                delta, finish = {"role": "assistant", "content": "PERSONAL_MCP_OK:" + identity}, "stop"
+            else:
+                name = next(item["function"]["name"] for item in body["tools"] if "personal_identity" in item["function"]["name"])
+                delta = {
+                    "role": "assistant",
+                    "tool_calls": [{"index": 0, "id": "personal-call", "type": "function", "function": {"name": name, "arguments": "{}"}}],
+                }
+                finish = "tool_calls"
+            common = {"id": "personal", "object": "chat.completion.chunk", "created": 1, "model": "deterministic-chat"}
+            send_openai_sse_response(
+                self,
+                [
+                    {**common, "choices": [{"index": 0, "delta": delta, "finish_reason": None}]},
+                    {
+                        **common,
+                        "choices": [{"index": 0, "delta": {}, "finish_reason": finish}],
+                        "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
+                    },
+                ],
+            )
+
+        def log_message(self, *_):
+            """测试不输出身份、凭据或请求正文。"""
+
+    with serve_replay_handler(Handler) as replay:
+        try:
+            await register_replay_provider(client, headers, base_url=replay, provider_id=provider)
+            # 一次性 HTTP fixture 直接登记可信连接，不修改部署的实验来源白名单。
+            # HTTPS 配置边界由 integration 单独证明；此处验证跨 worker 凭证与结果。
+            await db.execute(
+                "INSERT INTO mcp_servers (slug,name,url,transport,auth_mode,api_key_header,headers,enabled,created_by,updated_by) "
+                "VALUES ($1,'个人凭证验收',$2,'streamable_http',$3,$4,'{}',1,'fixture','fixture')",
+                slug,
+                replay + "/mcp",
+                mode,
+                "X-API-Key" if mode == "api_key" else None,
+            )
+            configured = await client.get("/api/system/mcp-servers/" + slug, headers=headers)
+            assert configured.status_code == 200
+            saved = await client.put(
+                "/api/system/mcp-servers/" + slug + "/connection",
+                headers=headers,
+                json={
+                    "secret": secret,
+                    "config_revision": configured.json()["data"]["config_revision"],
+                },
+            )
+            assert saved.status_code == 200
+            created = await client.post(
+                "/api/agent",
+                headers=headers,
+                json={
+                    "slug": agent,
+                    "name": "个人凭证验收",
+                    "backend_id": "ChatbotAgent",
+                    "config_json": {
+                        "context": {
+                            "model": provider + ":deterministic-chat",
+                            "mcps": [slug],
+                            "tools": [],
+                            "knowledges": [],
+                            "skills": [],
+                            "preload_skills": [],
+                        }
+                    },
+                },
+            )
+            assert created.status_code == 200, created.text
+            response = await client.post(
+                "/api/v1/agents/sessions",
+                headers={**headers, "Idempotency-Key": suffix},
+                json={
+                    "agent_id": agent,
+                    "title": "pytest-personal-" + suffix,
+                    "input": [public_message("读取本人身份")],
+                },
+            )
+            assert response.status_code == 201, response.text
+            accepted = await wait_for_consumed_input(client, headers, response.json()["yuxi"]["receipt"])
+            thread, turn_id = accepted["session_id"], accepted["turn_id"]
+            turn = await wait_for_terminal_turn(client, headers, thread, turn_id)
+            assert turn["status"] == "completed" and identity in output_text(turn["yuxi"]["output"])
+            run = turn["yuxi"]["result_run_id"]
+            assert run == accepted["run_id"] and observed == [identity]
+            record = await db.fetchrow("SELECT status,error_message FROM agent_runs WHERE id=$1", run)
+            assert record["status"] == "completed" and record["error_message"] is None
+        finally:
+            try:
+                if thread:
+                    await archive_public_thread(client, headers, thread, turn_id=turn_id)
+                await delete_agent(client, headers, agent)
+                await client.delete("/api/system/mcp-servers/" + slug, headers=headers)
+                await client.delete("/api/system/model-providers/" + provider, headers=headers)
+            finally:
+                await db.close()

backend/test/integration/services/test_app_initialization.py

@@ -71,7 +71,7 @@ async def test_fresh_app_and_rerun_preserve_admin_values_and_shared_files(isolat
     """完整初始化可重复执行，代码拥有的文件收敛且管理员字段保留。"""
     manager, engine, root = isolated_app
     await app_init.main()
-    assert await get_schema_versions(manager) == {"business": 1, "knowledge": 2}
+    assert await get_schema_versions(manager) == {"business": 2, "knowledge": 2}
     before = await persisted_facts(engine, root)
     async with engine.begin() as conn:
         await conn.execute(text("UPDATE config_options SET value='{\"admin\":\"preserve-this-value\"}', updated_by='admin-sentinel'"))
@@ -114,7 +114,7 @@ async def test_failed_shared_stage_preserves_schema_and_retry_finishes(isolated_
     monkeypatch.setattr(app_init, "init_builtin_skills", AsyncMock(side_effect=OSError("fixture storage unavailable")))
     with pytest.raises(OSError, match="fixture storage unavailable"):
         await app_init.main()
-    assert await get_schema_versions(manager) == {"business": 1, "knowledge": 2}
+    assert await get_schema_versions(manager) == {"business": 2, "knowledge": 2}
     async with engine.begin() as conn:
         assert await conn.scalar(text("SELECT count(*) FROM agents")) > 0
         await conn.execute(text("INSERT INTO knowledge_bases (kb_id,name,kb_type) VALUES ('retained','Retained','milvus')"))
@@ -166,7 +166,7 @@ async def run():
             assert "InvalidRequestError" in result.stderr and "Project" in result.stderr
         else:
             assert result.returncode == 0, result.stderr
-    assert await get_schema_versions(manager) == {"business": 1, "knowledge": 2}
+    assert await get_schema_versions(manager) == {"business": 2, "knowledge": 2}
     await persisted_facts(engine, root)
 
 
@@ -248,7 +248,7 @@ async def test_mineru_upgrade_persists_tier_without_overwriting_admin_fields(iso
         assert (
             await conn.execute(text("SELECT value, updated_by, updated_at FROM config_options WHERE key='mineru_ocr_host_opts'"))
         ).one() == row
-    assert await get_schema_versions(manager) == {"business": 1, "knowledge": 2}
+    assert await get_schema_versions(manager) == {"business": 2, "knowledge": 2}
 
 
 async def test_invalid_legacy_mineru_config_fails_without_replacing_value(isolated_app):

backend/test/integration/services/test_mcp_delegation_lifecycle.py

@@ -175,7 +175,15 @@ async def management_db():
             await db.commit()
         runtime = SimpleNamespace(uid=reader.uid, knowledges="all", run_id="independent-run")
         yield SimpleNamespace(
-            sessions=sessions, engine=engine, reader=reader, owner=owner, runtime=runtime, row=row, requests=requests, base_url=base_url
+            app=app,
+            sessions=sessions,
+            engine=engine,
+            reader=reader,
+            owner=owner,
+            runtime=runtime,
+            row=row,
+            requests=requests,
+            base_url=base_url,
         )
     finally:
         if http_server:
@@ -379,10 +387,15 @@ async def open_page(offset):
 
 @pytest.mark.parametrize("admin_enabled", [0, 1])
 async def test_management_http_lists_and_tests_delegated_server(isolated_mcp, admin_enabled):
-    """连接测试对停用服务仍严格建连；工具清单只列启用服务。"""
+    """普通用户不能测试委托服务，超管仍可检查停用服务的连接。"""
     context = isolated_mcp
     await update_server(context, enabled=admin_enabled, disabled_tools=["query_kb"])
     async with httpx.AsyncClient(base_url=context.base_url, trust_env=False) as client:
+        denied = await client.post("/api/system/mcp-servers/knowledge/test")
+        assert denied.status_code == 403, denied.text
+        assert not any(request["path"] == "/api/mcp" for request in context.requests)
+
+        context.app.dependency_overrides[get_required_user] = lambda: context.owner
         tested = await client.post("/api/system/mcp-servers/knowledge/test")
         listed = await client.get("/api/system/mcp-servers/knowledge/tools")
     assert tested.status_code == 200, tested.text