From f67b4d99b7c9526706aaba1d3be5ab6881fe507e Mon Sep 17 00:00:00 2001 From: yuxuanhui Date: Thu, 10 Sep 2026 10:26:44 +0800 Subject: [PATCH] feat(mcp): add WorldQuant authentication recovery tools --- backend/app/mcp_api/server.py | 4 +- backend/app/research_access/contracts.py | 8 +++ backend/app/research_access/service.py | 46 ++++++++++++++ backend/tests/test_mcp.py | 78 +++++++++++++++++++++++- docs/mcp-research.md | 8 ++- frontend/src/pages/MCPKeysPage.tsx | 2 +- 6 files changed, 142 insertions(+), 4 deletions(-) diff --git a/backend/app/mcp_api/server.py b/backend/app/mcp_api/server.py index 149c594..a702037 100644 --- a/backend/app/mcp_api/server.py +++ b/backend/app/mcp_api/server.py @@ -21,6 +21,8 @@ from ..research_access.service import ResearchAccess, ResearchError # Name, schema, business method, required scope, description. No generic arbitrary HTTP tool. TOOLS = { + "get_worldquant_connection": (c.ConnectionReference, "connection", "research:read", "读取 WorldQuant 连接状态及可选认证 job_id 的进度，不发起认证；人工验证在网页完成。"), + "authenticate_worldquant": (c.Authentication, "authenticate", "research:refresh", "使用服务端已保存凭据连接或重新认证 WorldQuant，返回 job_id；action=connect（默认）或人工验证后 verify。用 get_worldquant_connection 查询，不接收密码，不修改账户配置。"), "get_research_capabilities": (c.Empty, "capabilities", "research:read", "读取直接研究能力、完整设置 schema 和调度阻塞，不代表平台剩余额度。"), "search_catalog": (c.CatalogSearch, "catalog", "research:read", "分页查询指定范围的数据集或字段元数据；无缓存不等于无数据，不隐式刷新。"), "get_research_metadata": (c.Metadata, "metadata", "research:read", "读取范围、设置快照、算子定义或字段可用性；未知不认定通过。"), @@ -66,7 +68,7 @@ class MCPResearchServer: inputSchema=schema.model_json_schema(), annotations=types.ToolAnnotations( readOnlyHint=scope == "research:read", destructiveHint=method == "control", idempotentHint=method in {"submit", "control"} or scope == "research:read", - openWorldHint=method in {"refresh", "submit", "metadata", "check_self_correlation"})) + openWorldHint=method in {"refresh", "submit", "metadata", "check_self_correlation", "authenticate"})) for name, (schema, method, scope, description) in TOOLS.items() if scope in principal.scopes and "research:read" in principal.scopes]) diff --git a/backend/app/research_access/contracts.py b/backend/app/research_access/contracts.py index aa5b819..f57eb7d 100644 --- a/backend/app/research_access/contracts.py +++ b/backend/app/research_access/contracts.py @@ -18,6 +18,14 @@ class Empty(Contract): pass +class Authentication(Contract): + action: Literal["connect", "verify"] = "connect" + + +class ConnectionReference(Contract): + job_id: RunId | None = None + + class Page(Contract): limit: int = Field(default=25, ge=1, le=100) offset: int = Field(default=0, ge=0) diff --git a/backend/app/research_access/service.py b/backend/app/research_access/service.py index b5d8c94..4e893f2 100644 --- a/backend/app/research_access/service.py +++ b/backend/app/research_access/service.py @@ -17,6 +17,7 @@ from ..catalog.platform import platform_options, validate_platform_scope from ..catalog.research_metadata import ResearchMetadata, availability_key from ..catalog.service import Catalog from ..correlation import MIN_SAMPLES, THRESHOLD, WINDOW_YEARS +from ..jobs import AUTH_KINDS from ..models import Account, Alpha, BacktestItem, Job, JobItem, ResearchRequest, SimulationAttempt, now from ..research.serialization import encode_snapshot from ..research.workspace_contracts import FieldAvailabilityInput @@ -59,6 +60,51 @@ class ResearchAccess: "platform_check": False, }} + async def connection(self, args): + """Read bounded connection evidence; never return credentials or session cookies.""" + account = await self.db.get(Account, self.principal.account_id) + if not account or account.wq_user_id != self.principal.wq_user_id: + raise ResearchError("ACCOUNT_MISMATCH", "平台账户绑定已变化") + job = None + if args.job_id: + job = await self.db.get(Job, args.job_id) + if not job or job.kind not in AUTH_KINDS: + raise ResearchError("NOT_FOUND", "认证任务不存在") + return {"connection_status": account.connection_status, + "session_authenticated": self.client.authenticated, + "credentials_configured": bool(account.email and account.password_encrypted), + "requires_human_verification": account.connection_status == "verification_required", + "web_url": f"{self.public_origin}/", + "job": {"job_id": job.id, "kind": job.kind, "status": job.status} if job else None, + "next_step": "需要人工验证时在网页账户连接面板完成，再调用 authenticate_worldquant(action=verify)"} + + async def authenticate(self, args): + """Queue authentication using saved credentials, sharing the HTTP account lock. + + The caller commits the job and audit together before waking the runner. + Human challenges remain in the browser; no password or challenge URL is exposed. + """ + account = await self.db.scalar(select(Account).where( + Account.id == self.principal.account_id).with_for_update()) + if not account or account.wq_user_id != self.principal.wq_user_id: + raise ResearchError("ACCOUNT_MISMATCH", "平台账户绑定已变化") + if not account.email or not account.password_encrypted: + raise ResearchError("CREDENTIALS_NOT_CONFIGURED", "请先在网页保存 WorldQuant 账户配置") + job = await self.db.scalar(select(Job).where( + Job.kind.in_(AUTH_KINDS), Job.status.in_(("running", "queued")))) + if not job: + # Preserve a pending challenge instead of replacing its server-side session. + if args.action == "connect" and account.connection_status == "verification_required": + raise ResearchError("VERIFICATION_REQUIRED", "请在网页完成人工验证，再调用 action=verify") + job = Job(id=str(uuid4()), kind=args.action, payload={}, checkpoint={}) + self.db.add(job) + if args.action == "connect": + account.connection_status, account.connection_error = "connecting", None + await self.db.flush() + self.wake = "jobs" + return {"job_id": job.id, "status": job.status, "action": job.kind, + "read_with": "get_worldquant_connection", "web_url": f"{self.public_origin}/"} + async def catalog(self, args): data = await Catalog(self.db).search(args.filters, args.dataset_id) return {**data, "scope": args.filters.model_dump(include={"region", "universe", "delay", "instrument_type"}), diff --git a/backend/tests/test_mcp.py b/backend/tests/test_mcp.py index ec514d9..0aaaf84 100644 --- a/backend/tests/test_mcp.py +++ b/backend/tests/test_mcp.py @@ -162,7 +162,7 @@ async def test_official_sdk_client_and_error_contract(mcp_app): async with ClientSession(streams[0], streams[1]) as client: await client.initialize() listed = await client.list_tools() - assert len(listed.tools) == 13 + assert len(listed.tools) == 15 caps = await client.call_tool("get_research_capabilities", {}) assert caps.structured_content["max_candidates"] == 100 result = await client.call_tool("submit_backtests", submission()) @@ -383,3 +383,79 @@ async def test_self_correlation_read_only_scope_and_invalid_inputs(mcp_app): assert missing.is_error and missing.structured_content["error"]["code"] == "NOT_FOUND" async with mcp_app.state.sessions() as db: assert await db.scalar(select(func.count()).select_from(Job)) == 0 + + +async def test_worldquant_authentication_queue_and_recovery(mcp_app): + from app.models import Account, Job + + principal, _ = await credentials(mcp_app) + runner = mcp_app.state.runner + runner.client.disconnect() + async with mcp_app.state.sessions.begin() as db: + account = await db.get(Account, 1) + account.connection_status = "disconnected" + before = await invoke(mcp_app, principal, "get_worldquant_connection") + assert before["credentials_configured"] and not before["session_authenticated"] + runner.wake.clear() + started = await invoke(mcp_app, principal, "authenticate_worldquant") + assert started["status"] == "queued" and runner.wake.is_set() + again = await invoke(mcp_app, principal, "authenticate_worldquant") + assert again["job_id"] == started["job_id"] + await runner.execute(started["job_id"]) + done = await invoke(mcp_app, principal, "get_worldquant_connection", {"job_id": started["job_id"]}) + assert done["job"]["status"] == "completed" + assert done["connection_status"] == "connected" and done["session_authenticated"] + async with mcp_app.state.sessions() as db: + assert await db.scalar(select(func.count()).select_from(Job)) == 1 + audit = await db.scalar(select(MCPAudit).where(MCPAudit.tool == "authenticate_worldquant")) + assert audit.business_id == started["job_id"] + assert "synthetic-platform-secret" not in str(done) + assert "password" not in str(done) and "verification_url" not in str(done) + + +async def test_worldquant_authentication_permissions_and_challenge(mcp_app): + from fastapi import HTTPException + + from app.models import Account, Job + from app.worldquant import VerificationRequired + + readonly, _ = await credentials(mcp_app, {"research:read"}) + await invoke(mcp_app, readonly, "get_worldquant_connection") + with pytest.raises(HTTPException) as denied: + await mcp_app.state.mcp.invoke(readonly, "authenticate_worldquant", {}) + assert denied.value.status_code == 403 + principal, _ = await credentials(mcp_app) + invalid = await mcp_app.state.mcp.invoke(principal, "authenticate_worldquant", {"password": "untrusted"}) + assert invalid.is_error and invalid.structured_content["error"]["code"] == "INVALID_INPUT" + runner = mcp_app.state.runner + original = runner.client.authenticate + + async def challenge(*args, **kwargs): + runner.client.verification_url = "https://api.worldquantbrain.com/authentication/test-challenge" + raise VerificationRequired(runner.client.verification_url) + + runner.client.authenticate = challenge + started = await invoke(mcp_app, principal, "authenticate_worldquant") + await runner.execute(started["job_id"]) + waiting = await invoke(mcp_app, principal, "get_worldquant_connection", {"job_id": started["job_id"]}) + assert waiting["requires_human_verification"] and waiting["job"]["status"] == "waiting_auth" + blocked = await mcp_app.state.mcp.invoke(principal, "authenticate_worldquant", {}) + assert blocked.structured_content["error"]["code"] == "VERIFICATION_REQUIRED" + runner.client.authenticate = original + # Simulate completion of the human challenge; keep normal profile/identity verification. + async def verified(): + runner.client.verification_url = None + await original("synthetic@example.com", "synthetic-platform-secret", force=True) + runner.client.verify = verified + resumed = await invoke(mcp_app, principal, "authenticate_worldquant", {"action": "verify"}) + await runner.execute(resumed["job_id"]) + done = await invoke(mcp_app, principal, "get_worldquant_connection", {"job_id": resumed["job_id"]}) + assert done["connection_status"] == "connected" and done["job"]["status"] == "completed" + async with mcp_app.state.sessions.begin() as db: + account = await db.get(Account, 1) + account.password_encrypted = None + db.add(Job(id="unrelated", kind="pnl_refresh")) + missing = await mcp_app.state.mcp.invoke(principal, "authenticate_worldquant", {}) + assert missing.structured_content["error"]["code"] == "CREDENTIALS_NOT_CONFIGURED" + wrong = await mcp_app.state.mcp.invoke(principal, "get_worldquant_connection", {"job_id": "unrelated"}) + assert wrong.structured_content["error"]["code"] == "NOT_FOUND" diff --git a/docs/mcp-research.md b/docs/mcp-research.md index 960701a..6d3685c 100644 --- a/docs/mcp-research.md +++ b/docs/mcp-research.md @@ -34,7 +34,7 @@ python -m app.cli mcp-token-revoke TOKEN_ID | 权限 | 可调用能力 | | --- | --- | | research:read | 能力、数据目录、元数据、历史、运行、结果、证据、刷新及自相关任务查询、自相关结果读取 | -| research:refresh | 显式更新元数据及 PnL 缓存、发起本地自相关检查；同时要求 read | +| research:refresh | 显式更新元数据及 PnL 缓存、发起本地自相关检查、使用已保存凭据重新连接或继续认证；同时要求 read | | backtests:execute | 直接提交固定候选；同时要求 read | | backtests:control | 暂停、继续、停止、恢复采集；同时要求 read | @@ -46,6 +46,8 @@ python -m app.cli mcp-token-revoke TOKEN_ID | 工具 | 输入要点 | | --- | --- | +| get_worldquant_connection | `{job_id?}`；只读连接状态及认证任务状态，不返回密码、Cookie 或验证链接 | +| authenticate_worldquant | `{action?:"connect"或"verify"}`；默认 connect，使用已保存凭据异步认证，返回 job_id；要求 research:refresh | | get_research_capabilities | `{}`，含单次候选上限及完整候选 schema | | search_catalog | `{filters:{region,universe,delay,...},dataset_id?}`；省略 dataset_id 查数据集，提供则查字段 | | get_research_metadata | `{query:{kind,...}}`；kind 为 scopes/settings/operators/field_availability | @@ -64,6 +66,10 @@ metadata 的 operators 支持 q/category 和分页；settings 支持分页；fie 列表默认 25 项、最多 100 项；返回 total、offset、limit、has_more。快照证据按顶层 key/value 分页，嵌套内容完整保留；PnL 按日期过滤和分页。元数据仅提供字段与算子资料，不提供原始财务时间序列。 +## 连接恢复 + +遇到未连接时，调用 `authenticate_worldquant`，再以返回的 `job_id` 调用 `get_worldquant_connection`。受理不等于已连接；等待任务完成及连接状态为 connected。正在排队或执行的认证任务会复用。人工验证期间 connect 不替换验证会话：用户在返回的系统网页入口完成验证后调用 `{"action":"verify"}`。验证失败或缺少凭据时在网页账户面板处理。该入口不修改凭据，不签发或刷新 MCP PAT；PAT 失效仍需在管理页面处理。已有 research:refresh Key 可用，只读 Key 只能查询。 + ## 本地自相关检查 目标 Alpha 必须已导入；比较基准为本地已同步的同地区已提交 Alpha，排除自身。建议先在网页全量同步已提交 Alpha。MCP 不隐式导入 Alpha 或同步基准列表。 diff --git a/frontend/src/pages/MCPKeysPage.tsx b/frontend/src/pages/MCPKeysPage.tsx index 7397eeb..f225fd3 100644 --- a/frontend/src/pages/MCPKeysPage.tsx +++ b/frontend/src/pages/MCPKeysPage.tsx @@ -22,7 +22,7 @@ type TokenPage = { }; const scopes = [ ["research:read", "读取研究数据", "查询目录、历史、运行与结果"], - ["research:refresh", "刷新研究数据", "更新元数据和 PnL 缓存"], + ["research:refresh", "刷新研究数据", "更新缓存、检查自相关及恢复 WorldQuant 认证"], ["backtests:execute", "执行回测", "提交新的回测批次"], ["backtests:control", "控制回测", "暂停、继续、停止与恢复采集"], ] as const;