Compare commits

...

2 Commits

Author SHA1 Message Date
yuxuanhui be76b89e9d merge: integrate MCP research and key management
Deploy production / deploy (push) Successful in 1m5s
2026-09-09 16:18:40 +08:00
yuxuanhui 45238280e3 feat: add MCP research access and browser key management 2026-09-09 16:18:27 +08:00
47 changed files with 2642 additions and 44 deletions
+1
View File
@@ -24,4 +24,5 @@ jobs:
ADMIN_USERNAME: ${{ vars.ADMIN_USERNAME }}
DATABASE_NETWORK: ${{ vars.DATABASE_NETWORK }}
PUBLIC_ORIGIN: ${{ vars.PUBLIC_ORIGIN }}
MCP_ENABLED: ${{ vars.MCP_ENABLED }}
run: bash scripts/deploy-production.sh
@@ -0,0 +1,12 @@
# MCP Key 管理菜单与页面
Status: ready-for-agent
Type: task
实现 Cookie 管理接口、侧栏菜单、创建/单次明文/列表/撤销页面及必要测试;同步接入文档。
## Comments
- 用户已明确授权增加页面管理能力,替代首期仅 CLI 的范围限制。
- 已完成:`/api/v1/mcp-tokens` 管理接口,侧栏「系统管理 → MCP Key」,创建、单次明文、权限/有效期、分页列表、撤销,以及接入文档。
- 验证:后端全量 240 passed(79.24 秒),包含新增 2 项管理集成测试;新浏览器流程 1 passed(9.1 秒);前端构建、Ruff、受影响文件 Prettier 和 git diff --check 通过。
- 浏览器截图已检查,明文已遮罩。测试使用合成账户,无真实平台请求;未合并、推送或部署。
+7
View File
@@ -0,0 +1,7 @@
# MCP Key 网页管理
在已有 MCP 研究能力上增加系统管理菜单和页面,支持名称、1–365 天有效期、四种权限、默认只读、分页列表和撤销。复用原 PAT 签发与身份绑定,不新增迁移或改变 MCP 开关。
管理接口使用原 Cookie 会话及 X-WQ-Request / Origin 保护,PAT 不可管理 Key。明文提交成功后仅返回一次,不进入列表、日志、浏览器存储或助手上下文;关闭或离页销毁显示。列表显示有效、过期、撤销或绑定失效。服务关闭时可管理 Key,显示服务状态与连接地址。
验证覆盖认证隔离、请求保护、输入校验、默认权限、撤销即刻失效、分页无密文泄露及浏览器创建/离页/撤销。仅本地合成测试,不发布或创建生产令牌。
@@ -0,0 +1,16 @@
# 契约与增量迁移
Status: ready-for-agent
Blocked by: 无
按 ../spec.md 及会话批准计划实现并验证。
## Comments
- 已开始本地实现,禁止真实平台调用与部署。
## 实现结果
新增 typed contracts、三个持久模型、0010 增量迁移;SQLite 测试及 PostgreSQL 保留历史/回退重升级验证通过。
- 本地实现完成,未部署或使用真实平台。
@@ -0,0 +1,16 @@
# PAT、CLI 与 MCP 入口
Status: ready-for-agent
Blocked by: 01
按 ../spec.md 及会话批准计划实现并验证。
## Comments
- 已开始本地实现,禁止真实平台调用与部署。
## 实现结果
官方 SDK 2.2.0、ASGI lifespan、PAT 签发/列表/撤销、权限矩阵及 Host/Origin 校验已实现。官方客户端和容器代理验证通过。
- 本地实现完成,未部署或使用真实平台。
@@ -0,0 +1,16 @@
# 数据发现、刷新和历史检索
Status: ready-for-agent
Blocked by: 01,02
按 ../spec.md 及会话批准计划实现并验证。
## Comments
- 已开始本地实现,禁止真实平台调用与部署。
## 实现结果
数据目录、范围/算子/设置/字段可用性、显式缓存刷新、历史精确匹配及分页已实现。刷新错误有独立分页。
- 本地实现完成,未部署或使用真实平台。
@@ -0,0 +1,16 @@
# 直接提交幂等与控制
Status: ready-for-agent
Blocked by: 01,02,03
按 ../spec.md 及会话批准计划实现并验证。
## Comments
- 已开始本地实现,禁止真实平台调用与部署。
## 实现结果
直接候选提交、批内/历史重复策略、账户锁、成功请求幂等、控制重放及原队列唤醒已实现。PostgreSQL 并发和响应丢失测试通过。
- 本地实现完成,未部署或使用真实平台。
@@ -0,0 +1,16 @@
# 结果证据与网页留痕
Status: ready-for-agent
Blocked by: 04
按 ../spec.md 及会话批准计划实现并验证。
## Comments
- 已开始本地实现,禁止真实平台调用与部署。
## 实现结果
固定快照指标、非通过/未知检查、独立 PnL、证据分页、MCP 来源及运行定位已实现。浏览器相关四项测试通过。
- 本地实现完成,未部署或使用真实平台。
@@ -0,0 +1,16 @@
# 集成验收和使用文档
Status: ready-for-agent
Blocked by: 01,02,03,04,05
按 ../spec.md 及会话批准计划实现并验证。
## Comments
- 已开始本地实现,禁止真实平台调用与部署。
## 实现结果
已增加 MCP HTTP/官方客户端、PostgreSQL、浏览器及容器测试,使用文档已写入。最终回归结果见 ../verification.md。
- 本地实现完成,未部署或使用真实平台。
+22
View File
@@ -0,0 +1,22 @@
# MCP 研究能力
Status: ready-for-agent
按本会话确认的开发计划实现:远程 Streamable HTTP、个人访问令牌、11 个研究工具;复用现有业务模块和单进程回测队列。直接候选输入不要求研究资产。不上线、不签发生产令牌、不访问真实平台、不恢复定时研究。
## 契约
端点 /api/v1/mcp,默认关闭。权限 research:read、research:refresh、backtests:execute、backtests:control。PAT 绑定管理员与平台账户,默认 90 天,可撤销,仅保存哈希。提交最多 100 项,完整设置,重复默认拒绝,可明确 rerun。幂等范围为账户和操作;先幂等后去重,成功固定首次响应。停止不取消远程模拟,未知提交不重发。查询只读,刷新显式;摘要不隐藏失败、未知或缺失数据。
## 验收
认证与隔离、并发幂等、重复策略、重启与控制、固定结果快照、分页和 PnL、MCP 客户端与网页运行定位、迁移及回归。真实平台不在验收授权内。具体实现和实际检查记入 verification.md。
## 已实现结构
- MCP transport: backend/app/mcp_api/server.py;PAT: auth.py。独立 Bearer 身份、Host/Origin 校验、可撤销令牌 CLI。
- 共用业务: backend/app/research_access/contracts.py、service.py、queries.py。11 个有界工具,默认拒绝重复,成功请求原子固定响应。
- 迁移 0010 新增令牌、请求幂等和审计表。网页支持 #backtests?run_id=<ID>,来源显示与筛选 MCP。
- 配置 MCP_ENABLED 默认关闭,SDK 2.2.0 固定。首期没有 OAuth、远程取消、正式提交或定时研究。
完整输入输出、错误、权限和 CLI 说明见 ../../docs/mcp-research.md;实际检查见 verification.md。
+37
View File
@@ -0,0 +1,37 @@
# MCP 研究能力验证
日期:2026-09-09。仅本地合成平台;未部署、未签发生产令牌、未执行真实回测、未恢复定时研究。
## 实测
| 检查 | 结果 |
| --- | --- |
| `uv run pytest -q`(backend) | 238 passed,81.43 秒;包含新 MCP 9 项集成测试 |
| 官方 MCP ClientSession + Streamable HTTP | 初始化、11 工具发现、直接提交、结果查询及结构化错误通过;HTTP 边界连接真实应用,平台为 MockTransport |
| PAT | 缺失、过期、撤销、权限不足、账户变化、网页隔离、Host/Origin 拒绝与请求标识脱敏通过 |
| PAT CLI | 临时 SQLite 合成账户下,默认只读令牌签发、不含明文的列表和撤销通过 |
| PostgreSQL 17 隔离验收 | 并发提交/控制只创建一次,重启重放首次响应;0010 升级保留旧研究,回退/重升级通过 |
| Playwright backtests + research-navigation | 4 passed,42.6 秒;含实际 MCP 提交后同一运行的链接、刷新恢复和来源筛选 |
| 前端构建 | TypeScript + Vite 成功;现有第三方 lottie-web eval 提示仍存在 |
| 后端容器镜像 | Dockerfile.backend 成功构建,Linux Python 3.12 + MCP 2.2.0 |
| Caddy 容器代理 | 合成 PostgreSQL 后端经过现有 Caddyfile 验证:健康、无令牌401、有效令牌11工具、Origin拒绝、尾斜杠重定向通过 |
| 静态检查 | Ruff、受影响前端文件 Prettier、git diff --check 通过 |
PostgreSQL 使用独立 `wq_mcp_test` 数据库,不连接个人库;并发验收直接跨数据库事务调用共用业务模块,未依赖 MCP 进程内锁。响应丢失测试在服务完成后丢弃 HTTP 响应,重试只产生一个运行。
## 修正与核验
- Vite 原字符串代理改写 Host;改为显式 changeOrigin=false,与生产代理保持一致,未放宽 MCP 校验。
- SDK 2.x 模型属性使用 snake_case;协议线上的字段仍由 SDK 正确编码。
- pytest async fixture 的 MCP 生命周期放入同一任务进入/退出,避免 AnyIO cancel scope 跨任务退出。
- 审计请求标识保存哈希;管理员密码重置撤销 PAT;时区处理保留 aware datetime 的瞬时点;控制工具声明正确的副作用提示。
- 刷新任务错误独立分页;非通过检查和原始未知值完整保留,元数据刷新有30秒边界,不发布半成品。
- 独立业务核验已完成,未发现高置信的正确性缺陷;本地验收容器已清理。
## 实际边界
- 设置校验使用已缓存平台选项,缺快照返回 unknown;不宣称验证 FASTEXPR 语义或字段真实可用性。
- PnL 单独采集;snapshot 为脱敏历史快照,不是原始网络抓包。
- 平台远程取消、实际额度消耗、自动历史结果复用均不提供。
- Caddy 检查为本地 HTTP 容器链路,不替代公网 HTTPS、真实账户权限或真实客户端凭据配置验收。
- 容器代理检查使用构建后端镜像并只读挂载当前应用及合成测试工厂,测试代码由 .dockerignore 排除在发布镜像之外。
+6 -4
View File
@@ -62,7 +62,7 @@ docker compose ps
面板收起、切换会话和网络断开不会停止后端执行。刷新后从服务端历史与快照恢复,活动执行每 3 秒更新;不提供逐 token 续传。“停止生成”请求后端取消,再关闭前端接收。服务重启会将生成中的轮次标记为中断,不自动重放;待确认记录在重新登录后仍可处理,但重新检查版本。模型配置变更后,旧的待确认轮次需停止并重新预览。
模型不可用或未配置时,原有业务功能继续使用。API Key 仅加密存储于数据库,不返回浏览器;发送聊天时,相关本地业务结果会发送至你指定的模型服务。没有 MCP、知识检索、多 Agent 或平台属性回写。回测使用独立的固定集合确认,详见下文。
模型不可用或未配置时,原有业务功能继续使用。API Key 仅加密存储于数据库,不返回浏览器;发送聊天时,相关本地业务结果会发送至你指定的模型服务。MCP 研究入口独立于 Chatbox,见 [MCP 接入说明](docs/mcp-research.md);不提供知识检索、多 Agent 或平台属性回写。回测使用独立的固定集合确认,详见下文。
## Chatbox 研究到回测结果
@@ -111,6 +111,7 @@ docker compose -f compose.public.yaml logs --tail=100 web
| `WQ_EMAIL` / `WQ_PASSWORD` | WorldQuant 邮箱和密码,成对设置;本地 `.env`,生产 Gitea Secrets |
| `LOCAL_PORT` | 本机入口端口,默认 8080 |
| `DOMAIN` | 公网域名 |
| `MCP_ENABLED` | 启用远程 MCP 研究入口,默认 false;使用独立 PAT 认证 |
| `AI_REQUEST_LIMIT` | 每轮模型请求上限,默认 12 |
| `AI_TOOL_LIMIT` | 每轮工具执行上限,默认 12 |
| `AI_OUTPUT_TOKENS` | 每次模型输出上限,默认 4096 |
@@ -120,7 +121,7 @@ WorldQuant 凭据不再从 `account.json` 读取。进程环境变量优先于
WorldQuant 密码仅在后端解密。平台 Cookie 仅保存在后端内存,进程重启后重新认证。前端不保存密码或 Cookie 副本;日志与响应不输出平台认证正文。`.env` 不进入 Docker 构建上下文,应与数据库备份分别安全保管。丢失 `ENCRYPTION_KEY` 后须重新输入平台密码和模型 API Key;切勿在正常升级时重新生成它。
修改系统密码(同时撤销所有系统会话):
修改系统密码(同时撤销所有系统会话和 MCP 令牌):
```bash
docker compose exec backend python -m app.cli reset-password
@@ -222,20 +223,21 @@ WQ_DEV_API=http://127.0.0.1:18000 pnpm dev --port 5179
FastAPI 的 `/openapi.json` 与 `/docs` 可在后端开发端口访问;生产 Web 入口只代理 `/api/*`,不对外公开文档页面。
- `/api/v1/auth`:登录、退出、会话;除登录与健康检查外,业务接口都需要 Cookie。
- `/api/v1/auth`:登录、退出、会话;除登录、健康检查及独立 PAT 认证的 MCP 外,业务接口都需要 Cookie。
- `/api/v1/account`:偏好、加密凭据、连接/验证/断开/资料刷新。
- `/api/v1/alphas`:服务端筛选与排序、详情、本地研究记录、批量编辑、流式 CSV。
- `/api/v1/alphas/{id}/sources`:分页查看已保存回测的研究来源;Alpha 列表及 CSV 支持 `source`、`source_reference`、`research_id`、`backtest_run_id` 筛选。
- `/api/v1/alphas/{id}/pnl`:只读缓存;刷新通过 `pnl_refresh` 任务。
- `/api/v1/alphas/{id}/self-correlation`:读取本地检测结果;检测通过 `self_correlation` 任务。
- `/api/v1/sync-jobs`:创建任务立即返回 202 和 ID,查询、取消与重试。
- `/api/v1/mcp/`:可选 Streamable HTTP 研究入口,PAT 分权限认证;[工具与运维说明](docs/mcp-research.md)。
- `/api/v1/backtests`:候选草稿、不可变预览、异步启动、运行/结果/事件分页、调度配置、暂停/继续/停止/找回及重跑预览。
- `/api/v1/backtests/research-previews`:通过固定输入、表达式模板和字段绑定生成候选预览;沿用现有确认启动接口。
- `/api/v1/ai`:脱敏模型配置与测试、会话历史、SSE 执行、执行快照、取消及确认。新执行只接收 `request_id`、`message`、`context`;同一会话重复请求 ID 返回原运行,参数变化返回 409。
研究记录 PATCH 现在必须提供读取时的 `version`;批量编辑必须提供每个目标 ID 的 `versions` 映射。`0002` 迁移给旧研究记录设置初始版本 1,不修改其内容。版本冲突返回 409。
写请求需 `X-WQ-Request: 1`;浏览器跨站写入被拒绝。Alpha 平台快照、`research` 本地研究、`pnl_cache`、`self_correlations` 本地检测结果分开存储。`0005` 迁移只新增检测结果表。研究状态固定为 `inbox/candidate/optimizing/archived`;平台类型、语言、状态按原值显示。
网页写请求需 `X-WQ-Request: 1`;MCP 使用独立 Bearer 令牌且校验 Host/Origin;浏览器跨站写入被拒绝。Alpha 平台快照、`research` 本地研究、`pnl_cache`、`self_correlations` 本地检测结果分开存储。`0005` 迁移只新增检测结果表。研究状态固定为 `inbox/candidate/optimizing/archived`;平台类型、语言、状态按原值显示。
列表及导出支持 `submission=UNSUBMITTED|SUBMITTED`,平台状态缺失时不推断为已提交。`daily_sync` 必须提供分组及 `date_from` / `date_to`,每个 UTC 日期分别分页获取可见、隐藏记录;新建 `full_sync` 只同步已提交。旧的无分组全量任务保持原范围恢复。每页数据与检查点同事务提交,Alpha ID 幂等更新。失败任务保留进度,重试只处理剩余页或失败 ID。上游 `Retry-After` 等待可被取消。分页过程中平台记录移动可能造成重复或遗漏,通过 ID 去重和再次同步对应范围校正;单次没有查到不自动删除本地记录。
+50 -6
View File
@@ -4,11 +4,11 @@ import argparse
import asyncio
import getpass
from sqlalchemy import delete
from sqlalchemy import delete, update
from .config import Settings
from .db import create_database
from .models import Admin, LoginSession
from .models import Admin, LoginSession, MCPToken, now
from .security import password_hasher
@@ -21,13 +21,57 @@ async def reset_password():
admin = await db.get(Admin, 1)
admin.password_hash = password_hasher.hash(password)
await db.execute(delete(LoginSession))
await db.execute(update(MCPToken).where(MCPToken.revoked_at.is_(None)).values(revoked_at=now()))
await db.commit()
await engine.dispose()
print("Admin password updated; all system sessions revoked.")
print("Admin password updated; all system sessions and MCP tokens revoked.")
async def token_command(args):
import json
from sqlalchemy import select
from .mcp_api.auth import create_token
from .models import MCPToken, now
from .research.serialization import encode_snapshot
engine, sessions = create_database(Settings().database_url)
try:
async with sessions.begin() as db:
if args.command == "mcp-token-create":
row, secret = await create_token(db, args.name, args.scope, args.days)
result = {"id": row.id, "name": row.name, "scopes": row.scopes,
"expires_at": row.expires_at, "token": secret}
elif args.command == "mcp-token-revoke":
row = await db.get(MCPToken, args.token_id)
if not row:
raise ValueError("令牌不存在")
row.revoked_at = row.revoked_at or now()
result = {"id": row.id, "revoked": True}
else:
rows = list(await db.scalars(select(MCPToken).order_by(MCPToken.created_at.desc())))
result = [{k: getattr(row, k) for k in
("id", "name", "scopes", "created_at", "expires_at", "revoked_at")} for row in rows]
# Reveal only after the transaction has committed successfully.
print(json.dumps(encode_snapshot(result), ensure_ascii=False, indent=2))
finally:
await engine.dispose()
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=["reset-password"])
parser.parse_args()
asyncio.run(reset_password())
commands = parser.add_subparsers(dest="command", required=True)
commands.add_parser("reset-password")
create = commands.add_parser("mcp-token-create")
create.add_argument("--name", required=True)
create.add_argument("--scope", action="append", default=None)
create.add_argument("--days", type=int, default=90)
commands.add_parser("mcp-token-list")
revoke = commands.add_parser("mcp-token-revoke")
revoke.add_argument("token_id")
args = parser.parse_args()
try:
asyncio.run(reset_password() if args.command == "reset-password" else token_command(args))
except ValueError as exc:
parser.error(str(exc))
+1
View File
@@ -25,6 +25,7 @@ class Settings(BaseSettings):
request_timeout: float = 30
retry_attempts: int = Field(default=4, ge=1, le=8)
enable_runner: bool = True
mcp_enabled: bool = False
ai_request_limit: int = Field(default=12, ge=1, le=30)
ai_tool_limit: int = Field(default=12, ge=1, le=100)
ai_output_tokens: int = Field(default=4096, ge=128, le=32768)
+63 -2
View File
@@ -5,7 +5,7 @@ import csv
import io
import time
from collections import defaultdict
from contextlib import asynccontextmanager
from contextlib import AsyncExitStack, asynccontextmanager
from typing import Annotated
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request, Response
@@ -24,6 +24,7 @@ from .catalog.routes import router as catalog_router
from .config import Settings
from .db import create_database
from .jobs import AUTH_KINDS, Runner, create_job
from .mcp_api.token_routes import router as mcp_token_router
from .models import Account, Admin, BacktestConfig, Job, JobItem, LoginSession
from .research.routes import router as research_router
from .research.runtime import ResearchRuntime
@@ -93,6 +94,12 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None):
ai_runtime = AIRuntime(sessions, settings, runner, ai_model_factory)
research_runtime = ResearchRuntime(sessions, ai_runtime, runner)
mcp_runtime = None
if settings.mcp_enabled:
from .mcp_api.server import MCPResearchServer
mcp_runtime = MCPResearchServer(sessions, runner, settings)
@asynccontextmanager
async def lifespan(app):
async with sessions() as db:
@@ -104,6 +111,9 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None):
if settings.enable_runner:
await runner.start()
await research_runtime.start()
async with AsyncExitStack() as stack:
if mcp_runtime:
await stack.enter_async_context(mcp_runtime.server.session_manager.run())
yield
if settings.enable_runner:
await research_runtime.stop()
@@ -124,6 +134,7 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None):
app.state.settings = settings
app.state.ai = ai_runtime
app.state.research = research_runtime
app.state.mcp = mcp_runtime
login_failures = defaultdict(list)
@app.exception_handler(RequestValidationError)
@@ -139,7 +150,54 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None):
@app.middleware("http")
async def browser_security(request, call_next):
if request.method not in ("GET", "HEAD", "OPTIONS"):
is_mcp = request.url.path in ("/api/v1/mcp", "/api/v1/mcp/")
if is_mcp:
if not mcp_runtime:
return JSONResponse({"detail": "MCP 未启用"}, status_code=404)
from urllib.parse import urlsplit
from .mcp_api.auth import authenticate
from .mcp_api.server import TOOLS
if request.headers.get("host", "").lower() != urlsplit(settings.public_origin).netloc.lower():
return JSONResponse({"detail": "MCP Host 不被允许"}, status_code=403)
origin = request.headers.get("origin")
if origin and origin.rstrip("/") != settings.public_origin.rstrip("/"):
return JSONResponse({"detail": "MCP Origin 不被允许"}, status_code=403)
scheme, _, secret = request.headers.get("authorization", "").partition(" ")
if scheme.lower() != "bearer" or not secret or len(secret) > 256:
return JSONResponse({"detail": "需要 MCP Bearer 令牌"}, status_code=401,
headers={"WWW-Authenticate": "Bearer"})
try:
async with sessions() as db:
principal = await authenticate(db, secret)
except HTTPException as exc:
return JSONResponse({"detail": exc.detail}, status_code=exc.status_code,
headers={"WWW-Authenticate": "Bearer"})
request.state.mcp_principal = principal
if "research:read" not in principal.scopes:
return JSONResponse({"detail": "缺少读取权限"}, status_code=403)
if request.method == "POST":
body = bytearray()
async for chunk in request.stream():
body.extend(chunk)
if len(body) > 4 * 1024 * 1024:
return JSONResponse({"detail": "MCP 请求过大"}, status_code=413)
# BaseHTTPMiddleware replays cached bytes to the SDK; never log this payload.
request._body = bytes(body)
try:
import json
message = json.loads(body)
except (ValueError, UnicodeDecodeError):
return JSONResponse({"detail": "无效 JSON"}, status_code=400)
if isinstance(message, dict) and message.get("method") == "tools/call":
params = message.get("params")
tool = params.get("name") if isinstance(params, dict) else None
definition = TOOLS.get(tool) if isinstance(tool, str) else None
if definition and definition[2] not in principal.scopes:
return JSONResponse({"detail": "MCP 令牌缺少所需权限"}, status_code=403)
elif request.method not in ("GET", "HEAD", "OPTIONS"):
if request.headers.get("X-WQ-Request") != "1":
return JSONResponse({"detail": "缺少请求校验头"}, status_code=403)
origin = request.headers.get("Origin")
@@ -410,6 +468,9 @@ def create_app(settings=None, wq_client=None, ai_model_factory=None):
await notify_job(runner, "retry_job", result)
return result
if mcp_runtime:
app.mount("/api/v1/mcp", mcp_runtime.app)
app.include_router(mcp_token_router)
app.include_router(backtest_router)
app.include_router(api)
app.include_router(catalog_router)
+1
View File
@@ -0,0 +1 @@
"""Authenticated MCP transport; research behavior lives in research_access."""
+55
View File
@@ -0,0 +1,55 @@
"""Personal access tokens are isolated from browser and upstream credentials."""
import secrets
from dataclasses import dataclass
from datetime import timedelta, timezone
from uuid import uuid4
from fastapi import HTTPException
from sqlalchemy import select
from ..models import Account, Admin, MCPToken, now
from ..security import token_hash
SCOPES = frozenset({"research:read", "research:refresh", "backtests:execute", "backtests:control"})
@dataclass(frozen=True)
class Principal:
token_id: str
admin_id: int
account_id: int
wq_user_id: str
scopes: frozenset[str]
async def create_token(db, name, scopes=None, days=90):
"""Issue a token for the bound account; caller commits and reveals it once."""
scopes = set(scopes if scopes is not None else ["research:read"])
if not name.strip() or len(name) > 100 or not 1 <= days <= 365:
raise ValueError("名称须为 1–100 字,有效期须为 1–365 天")
if not scopes <= SCOPES or "research:read" not in scopes:
raise ValueError("权限无效;所有令牌必须包含 research:read")
account, admin = await db.get(Account, 1), await db.get(Admin, 1)
if not account or not account.wq_user_id or not admin:
raise ValueError("请先初始化系统并确认 WorldQuant 账户身份")
secret = "wqmcp_" + secrets.token_urlsafe(32)
row = MCPToken(
id=str(uuid4()), token_hash=token_hash(secret), name=name.strip(), admin_id=admin.id,
account_id=account.id, wq_user_id=account.wq_user_id, scopes=sorted(scopes),
expires_at=now() + timedelta(days=days),
)
db.add(row)
await db.flush()
return row, secret
async def authenticate(db, secret):
"""Validate every request, including current account binding; return no secrets."""
row = await db.scalar(select(MCPToken).where(MCPToken.token_hash == token_hash(secret)))
if not row or row.revoked_at or row.expires_at.replace(tzinfo=row.expires_at.tzinfo or timezone.utc) <= now():
raise HTTPException(401, "MCP 令牌无效或已过期")
account, admin = await db.get(Account, row.account_id), await db.get(Admin, row.admin_id)
if not account or not admin or account.id != 1 or account.wq_user_id != row.wq_user_id:
raise HTTPException(401, "MCP 令牌账户绑定已失效")
return Principal(row.id, row.admin_id, row.account_id, row.wq_user_id, frozenset(row.scopes))
+127
View File
@@ -0,0 +1,127 @@
"""MCP transport over shared research operations, with minimal durable audit evidence."""
import asyncio
import json
import time
from uuid import uuid4
import anyio
from fastapi import HTTPException
from mcp import types
from mcp.server.lowlevel import Server
from mcp.server.transport_security import TransportSecuritySettings
from pydantic import ValidationError
from ..alphas import sanitize
from ..backtests.contracts import fingerprint
from ..models import MCPAudit, now
from ..research.serialization import encode_snapshot
from ..research_access import contracts as c
from ..research_access.service import ResearchAccess, ResearchError
# Name, schema, business method, required scope, description. No generic arbitrary HTTP tool.
TOOLS = {
"get_research_capabilities": (c.Empty, "capabilities", "research:read", "读取直接研究能力、完整设置 schema 和调度阻塞,不代表平台剩余额度。"),
"search_catalog": (c.CatalogSearch, "catalog", "research:read", "分页查询指定范围的数据集或字段元数据;无缓存不等于无数据,不隐式刷新。"),
"get_research_metadata": (c.Metadata, "metadata", "research:read", "读取范围、设置快照、算子定义或字段可用性;未知不认定通过。"),
"refresh_research_data": (c.Refresh, "refresh", "research:refresh", "显式刷新目录、算子、设置、字段可用性或 PnL;不会创建模拟。任务返回 job_id。"),
"get_refresh_job": (c.JobReference, "refresh_job", "research:read", "查询研究刷新任务的状态和产物引用。"),
"search_backtests": (c.History, "history", "research:read", "分页查历史候选与固定设置;candidates 按完整输入精确匹配,不推断数学等价。"),
"submit_backtests": (c.Submit, "submit", "backtests:execute", "执行用户已授权的固定批次,自动留痕并立即返回运行 ID。每项必须完整设置;重复默认拒绝,rerun 明确重跑。不需要研究资产。"),
"get_backtest": (c.RunReference, "run", "research:read", "读取真实运行进度、提交数量和可选增量事件;受理不等于成功。"),
"get_backtest_results": (c.Results, "results", "research:read", "分页读取固定快照指标、全部非通过检查及三层状态;缺失指标不补零。"),
"get_backtest_artifact": (c.Artifact, "artifact", "research:read", "分页读取候选脱敏快照的顶层键值或独立采集的 PnL;缺缓存不自动刷新。"),
"control_backtest": (c.Control, "control", "backtests:control", "对已授权运行暂停、继续、停止或恢复采集;不远程取消、不重提未知模拟。需要版本和幂等键。"),
}
def tool_result(data, error=False):
data = encode_snapshot(data)
return types.CallToolResult(content=[types.TextContent(type="text", text=json.dumps(data, ensure_ascii=False))],
structuredContent=data, isError=error)
class MCPResearchServer:
def __init__(self, sessions, runner, settings):
self.sessions, self.runner, self.settings = sessions, runner, settings
# The existing deployment has one owner; this also gives SQLite test transactions a fair queue.
self.mutation_lock = asyncio.Lock()
self.server = Server("wq-alpha-research", version="1.0.0", on_list_tools=self.list_tools,
on_call_tool=self.call_tool,
instructions="自由探索,直接固定候选回测,无需先建研究资产。工具不安排定时研究;结果按运行 ID 查询。")
from urllib.parse import urlsplit
host = urlsplit(settings.public_origin).netloc
self.app = self.server.streamable_http_app(
streamable_http_path="/", stateless_http=True, json_response=True,
transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=True,
allowed_hosts=[host], allowed_origins=[settings.public_origin.rstrip("/")]),
)
async def list_tools(self, ctx, params):
principal = ctx.request.state.mcp_principal
return types.ListToolsResult(tools=[types.Tool(name=name, description=description,
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"}))
for name, (schema, method, scope, description) in TOOLS.items()
if scope in principal.scopes and "research:read" in principal.scopes])
async def call_tool(self, ctx, params):
principal = ctx.request.state.mcp_principal
return await self.invoke(principal, params.name, params.arguments or {}, str(ctx.request_id or uuid4()))
async def invoke(self, principal, name, arguments, request_id=None):
"""Invoke with a server-authenticated principal; atomic success audit and post-commit wake."""
started = time.monotonic()
request_id = request_id or str(uuid4())
entry = TOOLS.get(name)
if not entry:
return tool_result({"error": ResearchError("UNKNOWN_TOOL", "工具不存在").data}, True)
schema, method, scope, _ = entry
if "research:read" not in principal.scopes or scope not in principal.scopes:
raise HTTPException(403, "MCP 令牌缺少所需权限")
digest = fingerprint(arguments)
async with self.mutation_lock:
# Disconnect does not roll back an already accepted operation or lose its wake-up.
with anyio.CancelScope(shield=True):
async with self.sessions.begin() as db:
access = ResearchAccess(db, principal, self.runner.client, self.settings.public_origin)
code, error = "OK", False
try:
async with db.begin_nested():
args = schema.model_validate(arguments)
async with asyncio.timeout(30 if method in {"refresh", "metadata"} else None):
data = encode_snapshot(await getattr(access, method)(args))
data.setdefault("_meta", {"schema_version": 1, "observed_at": now().isoformat(),
"nulls": "null 表示来源未提供,不等于零", "source": "system"})
except ValidationError as exc:
code, error = "INVALID_INPUT", True
data = {"error": ResearchError(code, "; ".join(
f"{'.'.join(map(str, e['loc']))}: {e['msg']}" for e in exc.errors())).data}
except TimeoutError:
code, error = "UPSTREAM_TIMEOUT", True
data = {"error": ResearchError(code, "元数据读取或刷新超时,未发布新快照", retryable=True).data}
except ResearchError as exc:
code, error, data = exc.data["code"], True, {"error": exc.data}
except HTTPException as exc:
code = {404: "NOT_FOUND", 409: "CONFLICT", 422: "INVALID_INPUT", 429: "RATE_LIMITED", 502: "UPSTREAM_ERROR"}.get(exc.status_code, "REQUEST_FAILED")
error = True
data = {"error": ResearchError(code, str(sanitize(exc.detail)),
retryable=exc.status_code in {429, 502, 503},
retry_after=(exc.headers or {}).get("Retry-After")).data}
except Exception:
# Never expose SQL parameters, exception reprs or credentials in unexpected errors.
code, error = "INTERNAL_ERROR", True
data = {"error": ResearchError(code, "研究操作失败;可使用原幂等键重试或查询历史", retryable=True).data}
db.add(MCPAudit(id=str(uuid4()), token_id=principal.token_id, tool=name,
request_id=fingerprint({"request_id": request_id}), input_digest=digest,
business_id=data.get("backtest_run_id", data.get("job_id")),
result_code=code, elapsed_ms=int((time.monotonic()-started)*1000)))
if not error:
if access.wake == "backtests":
self.runner.backtests.wake.set()
elif access.wake == "jobs":
self.runner.wake.set()
return tool_result(data, error)
+82
View File
@@ -0,0 +1,82 @@
"""Cookie-authenticated PAT administration; MCP bearer tokens grant no access here."""
from datetime import timezone
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import func, select
from ..models import Account, MCPToken, now
from ..security import require_auth
from .auth import create_token
router = APIRouter(prefix="/api/v1/mcp-tokens", tags=["mcp-tokens"], dependencies=[Depends(require_auth)])
class TokenInput(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str = Field(min_length=1, max_length=100)
days: int = Field(default=90, ge=1, le=365, strict=True)
scopes: list[str] = Field(default_factory=lambda: ["research:read"], max_length=4)
def token_output(row, account):
"""Return public metadata only, including whether the current binding is usable."""
def timestamp(value):
return value.replace(tzinfo=value.tzinfo or timezone.utc) if value else None
expires = timestamp(row.expires_at)
status = (
"revoked" if row.revoked_at else
"expired" if expires <= now() else
"invalid_binding" if not account or account.wq_user_id != row.wq_user_id else
"active"
)
return {
"id": row.id, "name": row.name, "scopes": row.scopes,
"created_at": timestamp(row.created_at), "expires_at": expires,
"revoked_at": timestamp(row.revoked_at), "status": status,
}
@router.get("")
async def list_tokens(request: Request, limit: int = Query(25, ge=1, le=100), offset: int = Query(0, ge=0)):
async with request.app.state.sessions() as db:
account = await db.get(Account, 1)
owned = (MCPToken.admin_id == 1, MCPToken.account_id == 1)
total = await db.scalar(select(func.count()).select_from(MCPToken).where(*owned))
rows = await db.scalars(select(MCPToken).where(*owned).order_by(
MCPToken.created_at.desc(), MCPToken.id.desc()).offset(offset).limit(limit))
return {
"items": [token_output(row, account) for row in rows], "total": total,
"limit": limit, "offset": offset, "has_more": offset + limit < total,
"enabled": request.app.state.settings.mcp_enabled,
"endpoint": request.app.state.settings.public_origin.rstrip("/") + "/api/v1/mcp/",
"can_create": bool(account and account.wq_user_id),
}
@router.post("", status_code=201)
async def issue_token(body: TokenInput, request: Request):
# The existing single-admin browser session is the authority, never request-supplied IDs.
async with request.app.state.sessions.begin() as db:
try:
row, secret = await create_token(db, body.name, body.scopes, body.days)
except ValueError as exc:
raise HTTPException(422, str(exc)) from exc
result = token_output(row, await db.get(Account, 1))
# Do not expose the secret until the transaction successfully commits.
return {**result, "token": secret}
@router.post("/{token_id}/revoke")
async def revoke_token(token_id: str, request: Request):
async with request.app.state.sessions.begin() as db:
row = await db.scalar(select(MCPToken).where(
MCPToken.id == token_id, MCPToken.admin_id == 1, MCPToken.account_id == 1,
).with_for_update())
if not row:
raise HTTPException(404, "MCP Key 不存在")
row.revoked_at = row.revoked_at or now()
result = token_output(row, await db.get(Account, 1))
return result
+46
View File
@@ -494,3 +494,49 @@ class ResearchStepRun(Base):
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now)
__table_args__ = (UniqueConstraint("run_id", "node_id", "round"),)
class MCPToken(Base):
"""Revocable personal tokens; only the one-way digest is persisted."""
__tablename__ = "mcp_tokens"
id: Mapped[str] = mapped_column(String(36), primary_key=True)
token_hash: Mapped[str] = mapped_column(String(64), unique=True)
name: Mapped[str] = mapped_column(String(100))
admin_id: Mapped[int] = mapped_column(ForeignKey("admins.id"))
account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"))
wq_user_id: Mapped[str] = mapped_column(String(100))
scopes: Mapped[list] = mapped_column(JSON)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now)
expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
class ResearchRequest(Base):
"""Successful atomic operations survive retries and credential rotation."""
__tablename__ = "research_requests"
id: Mapped[str] = mapped_column(String(36), primary_key=True)
account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"))
operation: Mapped[str] = mapped_column(String(50))
idempotency_key: Mapped[str] = mapped_column(String(100))
digest: Mapped[str] = mapped_column(String(64))
business_id: Mapped[str] = mapped_column(String(36))
response: Mapped[dict] = mapped_column(JSON)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now)
__table_args__ = (UniqueConstraint("account_id", "operation", "idempotency_key"),)
class MCPAudit(Base):
"""Minimal call evidence, never raw arguments or authentication material."""
__tablename__ = "mcp_audits"
id: Mapped[str] = mapped_column(String(36), primary_key=True)
token_id: Mapped[str] = mapped_column(ForeignKey("mcp_tokens.id"), index=True)
tool: Mapped[str] = mapped_column(String(100))
request_id: Mapped[str] = mapped_column(String(100))
input_digest: Mapped[str] = mapped_column(String(64))
business_id: Mapped[str | None] = mapped_column(String(100))
result_code: Mapped[str] = mapped_column(String(60))
elapsed_ms: Mapped[int] = mapped_column(Integer)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=now, index=True)
+1
View File
@@ -0,0 +1 @@
"""Direct research interface shared by trusted application adapters."""
+170
View File
@@ -0,0 +1,170 @@
"""Bounded direct research inputs; unknown properties are rejected at the interface."""
from datetime import date, datetime
from typing import Annotated, Literal
from pydantic import Field, model_validator
from ..backtests.contracts import Candidate, SimulationSettings
from ..catalog.contracts import CatalogFilters, Scope
from ..schemas import Contract
Identifier = Annotated[str, Field(min_length=1, max_length=100)]
RunId = Annotated[str, Field(min_length=1, max_length=36)]
class Empty(Contract):
pass
class Page(Contract):
limit: int = Field(default=25, ge=1, le=100)
offset: int = Field(default=0, ge=0)
class CompleteSettings(SimulationSettings):
@model_validator(mode="before")
@classmethod
def complete(cls, value):
if isinstance(value, dict) and set(cls.model_fields) - value.keys():
raise ValueError("必须提供每项完整设置;先读取 get_research_capabilities")
return value
model_config = {"json_schema_extra": {"required": list(SimulationSettings.model_fields)}}
class DirectCandidate(Candidate):
settings: CompleteSettings
class Provenance(Contract):
reference: str | None = Field(default=None, max_length=200)
batch_id: str | None = Field(default=None, max_length=200)
hypothesis: str | None = Field(default=None, max_length=2000)
parent_run_id: RunId | None = None
class Submit(Contract):
name: str = Field(min_length=1, max_length=200)
candidates: list[DirectCandidate] = Field(min_length=1, max_length=100)
idempotency_key: Identifier
duplicate_policy: Literal["reject", "rerun"] = "reject"
source: Provenance = Field(default_factory=Provenance)
@model_validator(mode="after")
def unique_ids(self):
if len({c.client_item_id for c in self.candidates}) != len(self.candidates):
raise ValueError("client_item_id 必须唯一")
return self
class Control(Contract):
run_id: RunId
action: Literal["pause", "resume", "stop", "recover"]
expected_version: int = Field(ge=1)
idempotency_key: Identifier
class CatalogSearch(Contract):
filters: CatalogFilters
dataset_id: str | None = Field(default=None, min_length=1, max_length=200)
class Scopes(Contract):
kind: Literal["scopes"]
class SettingOptions(Page):
kind: Literal["settings"]
class Operators(Page):
kind: Literal["operators"]
q: str = Field(default="", max_length=300)
category: str | None = None
class Availability(Contract):
kind: Literal["field_availability"]
field_id: Identifier
scope: Scope
class Metadata(Contract):
query: Annotated[Scopes | SettingOptions | Operators | Availability, Field(discriminator="kind")]
class CatalogRefresh(Contract):
kind: Literal["catalog"]
scope: Scope
dataset_id: str | None = Field(default=None, min_length=1, max_length=200)
class OperatorsRefresh(Contract):
kind: Literal["operators"]
class SettingsRefresh(Contract):
kind: Literal["settings"]
class PnlRefresh(Contract):
kind: Literal["pnl"]
alpha_ids: list[Identifier] = Field(min_length=1, max_length=100)
class Refresh(Contract):
query: Annotated[
CatalogRefresh | OperatorsRefresh | SettingsRefresh | Availability | PnlRefresh,
Field(discriminator="kind"),
]
class JobReference(Page):
job_id: RunId
class History(Page):
source: str | None = Field(default=None, max_length=100)
reference: str | None = Field(default=None, max_length=200)
status: str | None = Field(default=None, max_length=30)
created_from: datetime | None = None
created_to: datetime | None = None
scope: Scope | None = None
q: str = Field(default="", max_length=300)
candidates: list[DirectCandidate] | None = Field(default=None, min_length=1, max_length=100)
@model_validator(mode="after")
def dates(self):
for value in (self.created_from, self.created_to):
if value and not value.tzinfo:
raise ValueError("时间须包含时区")
if self.created_from and self.created_to and self.created_from > self.created_to:
raise ValueError("起始时间不能晚于结束时间")
return self
class RunReference(Contract):
run_id: RunId
after: int | None = Field(default=None, ge=0)
event_limit: int = Field(default=25, ge=1, le=100)
class Results(Page):
run_id: RunId
item_ids: list[RunId] | None = Field(default=None, min_length=1, max_length=100)
class Artifact(Page):
item_id: RunId
kind: Literal["snapshot", "pnl"]
date_from: date | None = None
date_to: date | None = None
@model_validator(mode="after")
def dates(self):
if self.kind == "snapshot" and (self.date_from or self.date_to):
raise ValueError("日期筛选仅用于 PnL")
if self.date_from and self.date_to and self.date_from > self.date_to:
raise ValueError("起始日期不能晚于结束日期")
return self
+134
View File
@@ -0,0 +1,134 @@
"""Historical evidence reads, independent of transport and current Alpha refreshes."""
from collections import Counter
from sqlalchemy import func, select
from ..alphas import number, sanitize
from ..backtests.contracts import fingerprint
from ..models import BacktestItem, BacktestResult, BacktestRun, Pnl
from ..research.serialization import encode_snapshot
def page(items, total, limit, offset):
return {"items": items, "total": total, "limit": limit, "offset": offset,
"has_more": offset + len(items) < total}
def checks_summary(snapshot):
"""Preserve unknown check values; missing checks can never mean passed."""
checks = []
for section in ("is", "os"):
metrics = snapshot.get(section)
if isinstance(metrics, dict) and "checks" in metrics:
raw = metrics["checks"]
checks.extend({"section": section, "raw": c} for c in (raw if isinstance(raw, list) else [raw]))
if "checks" in snapshot:
raw = snapshot["checks"]
checks.extend({"section": "root", "raw": c} for c in (raw if isinstance(raw, list) else [raw]))
counts = Counter({key: 0 for key in ("PASS", "FAIL", "PENDING", "WARNING", "UNKNOWN")})
non_pass = []
for check in checks:
raw = check["raw"]
value = raw.get("result", raw.get("status")) if isinstance(raw, dict) else None
state = value if isinstance(value, str) and value in counts else "UNKNOWN"
counts[state] += 1
if state != "PASS":
non_pass.append({**check, "status": state})
return {"status": "unknown" if not checks else "reported", "counts": dict(counts),
"total": len(checks), "non_pass": non_pass}
def item_summary(item, result):
snapshot = sanitize(result.snapshot) if result else {}
metrics = {}
for section in ("is", "os"):
raw = snapshot.get(section)
raw = raw if isinstance(raw, dict) else {}
metrics[section] = {key: number(raw.get(key)) for key in
("sharpe", "fitness", "returns", "turnover", "margin", "drawdown")}
return encode_snapshot({
**{k: getattr(item, k) for k in (
"id", "run_id", "client_item_id", "expression", "settings", "attempt_id",
"platform_status", "collection_status", "persistence_status", "simulation_id", "alpha_id",
)},
"error": sanitize(item.error), "metrics": metrics,
"missing_metrics_reason": "来源未提供或非有限数字;null 不等于零",
"checks": checks_summary(snapshot),
"result": {"observed_at": result.observed_at, "complete": result.complete} if result else None,
"artifact_reference": {"item_id": item.id},
})
class EvidenceQueries:
def __init__(self, db):
self.db = db
async def history(self, args):
query = select(BacktestItem, BacktestResult, BacktestRun).join(
BacktestRun, BacktestRun.id == BacktestItem.run_id
).outerjoin(BacktestResult, BacktestResult.item_id == BacktestItem.id)
for key in ("source", "reference"):
value = getattr(args, key)
if value is not None:
query = query.where(BacktestRun.source["kind" if key == "source" else key].as_string() == value)
if args.status:
query = query.where(BacktestRun.status == args.status)
if args.created_from:
query = query.where(BacktestRun.created_at >= args.created_from)
if args.created_to:
query = query.where(BacktestRun.created_at <= args.created_to)
if args.scope:
for source, target in (("instrument_type", "instrumentType"), ("region", "region"), ("universe", "universe")):
query = query.where(BacktestItem.settings[target].as_string() == getattr(args.scope, source))
query = query.where(BacktestItem.settings["delay"].as_integer() == args.scope.delay)
if args.q:
escaped = args.q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
query = query.where(BacktestItem.expression.ilike(f"%{escaped}%", escape="\\"))
matches = {}
if args.candidates:
for c in args.candidates:
matches.setdefault(fingerprint(c.platform_input()), []).append(c.client_item_id)
query = query.where(BacktestItem.fingerprint.in_(matches))
total = await self.db.scalar(select(func.count()).select_from(query.subquery()))
rows = (await self.db.execute(query.order_by(BacktestRun.created_at.desc(), BacktestRun.id, BacktestItem.ordinal)
.limit(args.limit).offset(args.offset))).all()
items = [{**item_summary(i, r), "source": run.source, "run_status": run.status,
"created_at": run.created_at, "matched_candidates": matches.get(i.fingerprint, []),
"match_type": "exact_input" if args.candidates else "filter"} for i, r, run in rows]
return encode_snapshot(page(items, total, args.limit, args.offset))
async def results(self, args):
query = select(BacktestItem, BacktestResult).outerjoin(
BacktestResult, BacktestResult.item_id == BacktestItem.id
).where(BacktestItem.run_id == args.run_id)
if args.item_ids:
query = query.where(BacktestItem.id.in_(args.item_ids))
total = await self.db.scalar(select(func.count()).select_from(query.subquery()))
rows = (await self.db.execute(query.order_by(BacktestItem.ordinal).limit(args.limit).offset(args.offset))).all()
return {"backtest_run_id": args.run_id, **page([item_summary(i, r) for i, r in rows], total, args.limit, args.offset)}
async def artifact(self, args):
from .service import ResearchError
item = await self.db.get(BacktestItem, args.item_id)
if not item:
raise ResearchError("NOT_FOUND", "候选不存在")
result = await self.db.get(BacktestResult, item.id)
if args.kind == "snapshot":
# Top-level entries retain complete nested values; no hidden string/list truncation.
entries = [{"key": k, "value": v} for k, v in sanitize(result.snapshot).items()] if result else []
return encode_snapshot({"item_id": item.id, "kind": args.kind,
"status": "available" if result else "not_available",
"observed_at": result.observed_at if result else None,
"complete": result.complete if result else False,
**page(entries[args.offset:args.offset + args.limit], len(entries), args.limit, args.offset)})
pnl = await self.db.get(Pnl, item.alpha_id) if item.alpha_id else None
points = pnl.points if pnl else []
points = [p for p in points if
(not args.date_from or p["date"][:10] >= args.date_from.isoformat()) and
(not args.date_to or p["date"][:10] <= args.date_to.isoformat())]
return encode_snapshot({"item_id": item.id, "alpha_id": item.alpha_id, "kind": args.kind,
"status": "available" if pnl else "not_cached", "fetched_at": pnl.fetched_at if pnl else None,
"units": "供应商原始累计值;未提供货币或规模单位",
**page(points[args.offset:args.offset + args.limit], len(points), args.limit, args.offset)})
+222
View File
@@ -0,0 +1,222 @@
"""Direct research operations; caller owns authorization, transaction and wake-up.
The account row serializes mutations with existing HTTP starts. Request records,
previews, runs and control events commit together; failed validation consumes no key.
"""
from collections import Counter
from uuid import uuid4
from sqlalchemy import func, select
from ..backtests.contracts import ControlInput, DraftInput, PreviewInput, Source, StartInput, fingerprint
from ..backtests.service import Backtests
from ..business import Business
from ..catalog.contracts import CatalogJobInput
from ..catalog.platform import platform_options, validate_platform_scope
from ..catalog.research_metadata import ResearchMetadata, availability_key
from ..catalog.service import Catalog
from ..models import Account, Alpha, BacktestItem, Job, JobItem, ResearchRequest, SimulationAttempt, now
from ..research.serialization import encode_snapshot
from ..research.workspace_contracts import FieldAvailabilityInput
from ..schemas import JobInput
from .contracts import DirectCandidate, History
from .queries import EvidenceQueries, page
class ResearchError(Exception):
def __init__(self, code, message, *, retryable=False, retry_after=None, affected_items=None):
super().__init__(message)
self.data = {"code": code, "message": message, "retryable": retryable,
"retry_after": retry_after, "affected_items": affected_items or []}
class ResearchAccess:
def __init__(self, db, principal, client, public_origin):
self.db, self.principal, self.client = db, principal, client
self.public_origin = public_origin.rstrip("/")
self.backtests = Backtests(db)
self.business = Business(db)
self.evidence = EvidenceQueries(db)
self.wake = None
def run_url(self, run_id):
return f"{self.public_origin}/#backtests?run_id={run_id}"
async def capabilities(self, args):
return {**await self.backtests.capabilities(), "max_candidates": 100,
"settings_schema": DirectCandidate.model_json_schema(),
"confirmation": "调用者须已获本批执行授权;直接提交后返回稳定运行 ID",
"duplicate_policies": ["reject", "rerun"], "permissions": sorted(self.principal.scopes),
"metadata_only": True, "actual_platform_allowance": None}
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"}),
"status": "available" if data["collection_version"] else "not_cached",
"has_more": data["offset"] + len(data["items"]) < data["total"]}
async def metadata(self, args):
q = args.query
metadata = ResearchMetadata(self.db)
if q.kind == "scopes":
return {"source": "worldquant_platform", **await platform_options(self.client)}
if q.kind == "operators":
data = await metadata.operators(q.q, q.category, limit=q.limit, offset=q.offset)
return {**data, "status": "available" if data["fetched_at"] else "not_cached",
"has_more": q.offset + len(data["items"]) < data["total"]}
if q.kind == "settings":
data = await metadata.get("settings")
items = data["content"].get("items", [])
return {"status": "available" if data["fetched_at"] else "not_cached",
"fetched_at": data["fetched_at"], **page(items[q.offset:q.offset+q.limit], len(items), q.limit, q.offset)}
data = await metadata.get(availability_key(q.field_id, q.scope))
return {**data, "status": data["content"].get("status", "unknown")}
async def refresh(self, args):
q = args.query
metadata = ResearchMetadata(self.db, self.client)
if q.kind == "catalog":
await validate_platform_scope(self.client, q.scope)
job = await Catalog(self.db).create_job(CatalogJobInput(scope=q.scope, dataset_id=q.dataset_id))
self.wake = "jobs"
return {"job_id": job.id, "status": job.status}
if q.kind == "pnl":
ids = sorted(set(q.alpha_ids))
existing = set(await self.db.scalars(select(Alpha.id).where(Alpha.id.in_(ids))))
if existing != set(ids):
raise ResearchError("NOT_FOUND", "部分 Alpha 尚未同步", affected_items=sorted(set(ids)-existing))
job = await self.business.create_sync_job(JobInput(kind="pnl_refresh", alpha_ids=ids))
self.wake = "jobs"
return {"job_id": job["id"], "status": job["status"]}
if q.kind == "operators":
data = await metadata.refresh_operators()
elif q.kind == "settings":
data = await metadata.refresh_settings()
else:
data = await metadata.refresh_availability(FieldAvailabilityInput(field_id=q.field_id, scope=q.scope))
# Refresh acknowledgment is bounded; complete content is available through paged reads.
return {"status": "completed", "key": data["key"], "fetched_at": data["fetched_at"],
"read_with": "get_research_metadata"}
async def refresh_job(self, args):
job = await self.db.get(Job, args.job_id)
if not job or job.kind not in {"catalog_sync", "field_sync", "pnl_refresh"}:
raise ResearchError("NOT_FOUND", "研究刷新任务不存在")
result = await self.business.get_job_status(args.job_id)
query = select(JobItem).where(JobItem.job_id == job.id, JobItem.error.is_not(None))
total = await self.db.scalar(select(func.count()).select_from(query.subquery()))
errors = list(await self.db.scalars(query.order_by(JobItem.alpha_id).limit(args.limit).offset(args.offset)))
result.pop("errors", None)
return {**result, "job_id": job.id, "artifact_reference": job.payload,
"errors": page([{"alpha_id": e.alpha_id, "error": e.error} for e in errors], total, args.limit, args.offset)}
async def history(self, args):
return await self.evidence.history(args)
async def previous(self, operation, args):
# PostgreSQL row lock is shared with HTTP start and catalog/job creation.
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", "平台账户绑定已变化")
digest = fingerprint(args.model_dump(mode="json", exclude={"idempotency_key"}))
row = await self.db.scalar(select(ResearchRequest).where(
ResearchRequest.account_id == account.id, ResearchRequest.operation == operation,
ResearchRequest.idempotency_key == args.idempotency_key))
if row and row.digest != digest:
raise ResearchError("IDEMPOTENCY_CONFLICT", "幂等键已用于不同内容")
return row, digest
async def remember(self, operation, args, digest, result):
result["_meta"] = {"schema_version": 1, "observed_at": now().isoformat(), "source": "system"}
self.db.add(ResearchRequest(id=str(uuid4()), account_id=self.principal.account_id,
operation=operation, idempotency_key=args.idempotency_key, digest=digest,
business_id=result["backtest_run_id"], response=encode_snapshot(result)))
await self.db.flush()
self.wake = "backtests"
return result
async def validate_settings(self, candidates):
snapshot = await ResearchMetadata(self.db).get("settings")
options = snapshot["content"].get("items", [])
if not snapshot["fetched_at"] or not options:
return {"settings_validation": "unknown", "reason": "设置快照未缓存;未验证平台组合", "field_validation": "unknown"}
invalid = []
for c in candidates:
s = c.settings
matches = [r for r in options if all(r.get(k) == v for k, v in {
"instrument_type": s.instrumentType, "region": s.region, "universe": s.universe, "delay": s.delay}.items())]
if not matches or all(r.get("neutralizations") and s.neutralization not in r["neutralizations"] for r in matches):
invalid.append(c.client_item_id)
if invalid:
raise ResearchError("UNSUPPORTED_SETTINGS", "已缓存平台设置不支持这些组合;可显式刷新后重试", affected_items=invalid)
return {"settings_validation": "cached", "fetched_at": snapshot["fetched_at"], "field_validation": "unknown"}
async def submit(self, args):
previous, digest = await self.previous("submit_backtests", args)
if previous:
return previous.response
seen, within = {}, []
for c in args.candidates:
h = fingerprint(c.platform_input())
if h in seen:
within.append({"client_item_id": c.client_item_id, "duplicate_of": seen[h]})
seen[h] = c.client_item_id
history = await self.evidence.history(History(candidates=args.candidates, limit=100))
if args.duplicate_policy == "reject" and (within or history["total"]):
raise ResearchError("DUPLICATE_INPUT", "发现完整输入重复;未创建运行。重跑须明确 duplicate_policy=rerun",
affected_items={"within_batch": within, "history": history, "read_with": "search_backtests"})
validation = await self.validate_settings(args.candidates)
if args.source.parent_run_id:
await self.backtests.run(args.source.parent_run_id)
source = Source(kind="mcp", **args.source.model_dump())
provenance = {"mcp_token_id": self.principal.token_id, "admin_id": self.principal.admin_id}
# preserve_source prevents the Chatbox-specific generating context rewriting MCP provenance.
backtests = Backtests(self.db, provenance)
preview = await backtests.preview(PreviewInput(inline=DraftInput(
name=args.name, source=source, candidates=args.candidates)), preserve_source=True)
result = await backtests.start(StartInput(preview_id=preview["preview_id"],
idempotency_key="mcp-" + str(uuid4())))
result = {**result, "input_digest": digest, "batch_count": preview["batch_count"],
"duplicates": {"within_batch": within, "historical_matches": history["total"]},
"validation": validation, "web_url": self.run_url(result["backtest_run_id"])}
return await self.remember("submit_backtests", args, digest, result)
async def run(self, args):
result = await self.backtests.run(args.run_id)
attempts = list(await self.db.scalars(select(SimulationAttempt).where(SimulationAttempt.run_id == args.run_id)))
result["submission_counts"] = {
"candidates": result["total"], "attempts": len(attempts),
"post_requests": sum(a.submit_count for a in attempts),
"confirmed_accepted_candidates": sum(len(a.payload) for a in attempts if a.progress_url),
"unknown_acceptance_candidates": sum(len(a.payload) for a in attempts if a.error_code == "submission_unknown" or (a.state == "submitting" and not a.progress_url)),
"actual_platform_consumption": None,
}
if args.after is not None:
result["events"] = await self.backtests.events(args.run_id, args.after, args.event_limit)
return {**result, "web_url": self.run_url(args.run_id)}
async def results(self, args):
await self.backtests.run(args.run_id)
if args.item_ids:
found = set(await self.db.scalars(select(BacktestItem.id).where(
BacktestItem.run_id == args.run_id, BacktestItem.id.in_(args.item_ids))))
if found != set(args.item_ids):
raise ResearchError("NOT_FOUND", "部分候选不属于此运行")
return await self.evidence.results(args)
async def artifact(self, args):
return await self.evidence.artifact(args)
async def control(self, args):
previous, digest = await self.previous("control_backtest", args)
if previous:
return previous.response
before = await self.backtests.run(args.run_id)
states = list(await self.db.scalars(select(SimulationAttempt.state).where(SimulationAttempt.run_id == args.run_id)))
result = await self.backtests.control(args.run_id, ControlInput(action=args.action, version=args.expected_version))
result["impact"] = {"remote_cancelled": False, "attempts_before": dict(Counter(states)),
"indefinite_account_block_cleared": args.action == "resume" and bool(before["scheduler"]["blocked_reason"])
and before["scheduler"]["blocked_until"] is None,
"note": "暂停/停止仅阻止后续提交;已提交模拟继续采集。recover 不重新提交。"}
return await self.remember("control_backtest", args, digest, result)
@@ -0,0 +1,58 @@
"""Personal tokens, atomic research requests and minimal MCP audit evidence."""
import sqlalchemy as sa
from alembic import op
revision = "0010"
down_revision = "0009"
branch_labels = None
depends_on = None
def upgrade():
op.create_table(
"mcp_tokens",
sa.Column("id", sa.String(36), primary_key=True),
sa.Column("token_hash", sa.String(64), nullable=False, unique=True),
sa.Column("name", sa.String(100), nullable=False),
sa.Column("admin_id", sa.Integer(), sa.ForeignKey("admins.id"), nullable=False),
sa.Column("account_id", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=False),
sa.Column("wq_user_id", sa.String(100), nullable=False),
sa.Column("scopes", sa.JSON(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("revoked_at", sa.DateTime(timezone=True)),
)
op.create_index("ix_mcp_tokens_expires_at", "mcp_tokens", ["expires_at"])
op.create_table(
"research_requests",
sa.Column("id", sa.String(36), primary_key=True),
sa.Column("account_id", sa.Integer(), sa.ForeignKey("accounts.id"), nullable=False),
sa.Column("operation", sa.String(50), nullable=False),
sa.Column("idempotency_key", sa.String(100), nullable=False),
sa.Column("digest", sa.String(64), nullable=False),
sa.Column("business_id", sa.String(36), nullable=False),
sa.Column("response", sa.JSON(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint("account_id", "operation", "idempotency_key"),
)
op.create_table(
"mcp_audits",
sa.Column("id", sa.String(36), primary_key=True),
sa.Column("token_id", sa.String(36), sa.ForeignKey("mcp_tokens.id"), nullable=False),
sa.Column("tool", sa.String(100), nullable=False),
sa.Column("request_id", sa.String(100), nullable=False),
sa.Column("input_digest", sa.String(64), nullable=False),
sa.Column("business_id", sa.String(100)),
sa.Column("result_code", sa.String(60), nullable=False),
sa.Column("elapsed_ms", sa.Integer(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_mcp_audits_token_id", "mcp_audits", ["token_id"])
op.create_index("ix_mcp_audits_created_at", "mcp_audits", ["created_at"])
def downgrade():
op.drop_table("mcp_audits")
op.drop_table("research_requests")
op.drop_table("mcp_tokens")
+1
View File
@@ -14,6 +14,7 @@ dependencies = [
"cryptography>=44,<50",
"argon2-cffi>=23.1,<26",
"pydantic-ai-slim[openai]==1.97.0",
"mcp>=2,<3",
]
[dependency-groups]
+13
View File
@@ -81,6 +81,7 @@ def create_test_app():
_env_file=None,
database_url=f"sqlite+aiosqlite:///{temporary.name}/test.db",
admin_password=TEST_PASSWORD,
mcp_enabled=True,
encryption_key=Fernet.generate_key().decode(),
public_origin=os.environ.get("BROWSER_TEST_ORIGIN", "http://127.0.0.1:5179"),
)
@@ -206,6 +207,18 @@ def create_test_app():
application = create_app(
settings, WqClient(settings, transport=httpx.MockTransport(upstream)), ai_model_factory=fake_model
)
# Test-only fixture, excluded from production images by .dockerignore.
from fastapi import Depends
from app.mcp_api.auth import SCOPES, create_token
from app.security import require_auth
@application.post("/api/v1/test/mcp-token", dependencies=[Depends(require_auth)])
async def synthetic_token():
async with application.state.sessions.begin() as db:
_, token = await create_token(db, "browser-fixture", SCOPES)
return {"token": token}
original_lifespan = application.router.lifespan_context
@asynccontextmanager
+44
View File
@@ -0,0 +1,44 @@
"""Synthetic PostgreSQL-backed server for container proxy acceptance only."""
from contextlib import asynccontextmanager
import httpx
from fastapi import Depends
from app.config import Settings
from app.main import create_app
from app.mcp_api.auth import SCOPES, create_token
from app.models import Account
from app.security import cipher, require_auth
from app.worldquant import WqClient
from tests.backtest_fake import Platform
def create_test_app():
settings = Settings(_env_file=None, enable_runner=False, mcp_enabled=True)
if not settings.database_url.endswith("/wq_mcp_test"):
raise RuntimeError("Disposable test database required")
platform = Platform()
app = create_app(settings, WqClient(settings, transport=httpx.MockTransport(platform)))
original = app.router.lifespan_context
@asynccontextmanager
async def lifespan(application):
async with original(application):
async with app.state.sessions.begin() as db:
account = await db.get(Account, 1)
account.email = "synthetic@example.com"
account.wq_user_id = "TEST_USER"
account.connection_status = "connected"
account.password_encrypted = cipher(settings).encrypt(platform.secret.encode()).decode()
yield
app.router.lifespan_context = lifespan
@app.post("/api/v1/test/mcp-token", dependencies=[Depends(require_auth)])
async def issue_fixture():
async with app.state.sessions.begin() as db:
_, token = await create_token(db, "container-fixture", SCOPES)
return {"token": token}
return app
+97
View File
@@ -0,0 +1,97 @@
"""Disposable PostgreSQL migration/concurrency acceptance; synthetic platform only.
MCP_TEST_DATABASE_URL must identify a local database named wq_mcp_test.
"""
import asyncio
import os
from urllib.parse import urlsplit
import httpx
from alembic import command
from alembic.config import Config
from cryptography.fernet import Fernet
from sqlalchemy import func, select
from app.alphas import upsert_alpha
from app.config import Settings
from app.db import create_database
from app.main import create_app
from app.mcp_api.auth import SCOPES, authenticate, create_token
from app.models import BacktestEvent, BacktestPreview, BacktestRun, Research, ResearchRequest
from app.research_access.contracts import Control, Submit
from app.research_access.service import ResearchAccess, ResearchError
from app.worldquant import WqClient
from tests.backtest_fake import Platform
from tests.test_backtests import candidate, setup
async def seed(settings):
engine, sessions = create_database(settings.database_url)
async with sessions.begin() as db:
await upsert_alpha(db, {"id": "legacy", "regular": {"code": "rank(close)"}})
(await db.get(Research, "legacy")).note = "keep historical note"
await engine.dispose()
async def verify(settings):
app = create_app(settings, WqClient(settings, transport=httpx.MockTransport(Platform())))
async with app.router.lifespan_context(app):
await setup(app)
async with app.state.sessions.begin() as db:
_, secret = await create_token(db, "synthetic", SCOPES)
principal = await authenticate(db, secret)
args = Submit(name="concurrent", candidates=[candidate()], idempotency_key="same")
async def submit(body):
async with app.state.sessions.begin() as db:
return await ResearchAccess(db, principal, app.state.runner.client, settings.public_origin).submit(body)
a, b = await asyncio.gather(submit(args), submit(args))
assert a == b
rid = a["backtest_run_id"]
async with app.state.sessions() as db:
assert await db.scalar(select(func.count()).select_from(BacktestRun)) == 1
assert await db.scalar(select(func.count()).select_from(BacktestPreview)) == 1
assert await db.scalar(select(func.count()).select_from(ResearchRequest)) == 1
assert (await db.get(Research, "legacy")).note == "keep historical note"
try:
await submit(args.model_copy(update={"name": "different"}))
except ResearchError as exc:
assert exc.data["code"] == "IDEMPOTENCY_CONFLICT"
else:
raise AssertionError("different payload accepted")
control = Control(run_id=rid, action="pause", expected_version=1, idempotency_key="pause")
async def pause():
async with app.state.sessions.begin() as db:
return await ResearchAccess(db, principal, app.state.runner.client, settings.public_origin).control(control)
x, y = await asyncio.gather(pause(), pause())
assert x == y
async with app.state.sessions() as db:
assert await db.scalar(select(func.count()).select_from(BacktestEvent).where(BacktestEvent.kind == "control")) == 1
# A fresh process-equivalent application returns the original response after restart.
replacement = create_app(settings, WqClient(settings, transport=httpx.MockTransport(Platform())))
async with replacement.router.lifespan_context(replacement):
async with replacement.state.sessions.begin() as db:
principal = await authenticate(db, secret)
replay = await ResearchAccess(db, principal, replacement.state.runner.client, settings.public_origin).submit(args)
assert replay == a
print("PostgreSQL: migration preserves history; concurrent submit/control and restart replay passed")
if __name__ == "__main__":
url = os.environ["MCP_TEST_DATABASE_URL"]
parsed = urlsplit(url)
if parsed.hostname not in {"127.0.0.1", "localhost"} or parsed.path != "/wq_mcp_test":
raise SystemExit("Refusing a non-local/non-disposable database")
key = Fernet.generate_key().decode()
os.environ.update(DATABASE_URL=url, ADMIN_PASSWORD="synthetic-admin-only", ENCRYPTION_KEY=key,
WQ_EMAIL="", WQ_PASSWORD="")
settings = Settings(_env_file=None, database_url=url, admin_password="synthetic-admin-only",
encryption_key=key, enable_runner=False, mcp_enabled=True, public_origin="http://testserver")
config = Config("alembic.ini")
command.upgrade(config, "0009")
asyncio.run(seed(settings))
command.upgrade(config, "head")
asyncio.run(verify(settings))
command.downgrade(config, "0009")
command.upgrade(config, "head")
print("PostgreSQL: additive migration downgrade/re-upgrade passed")
+280
View File
@@ -0,0 +1,280 @@
"""MCP integration uses real database/transport and a synthetic WorldQuant only."""
import asyncio
from datetime import timedelta
import httpx
import pytest
from sqlalchemy import func, select
from app.mcp_api.auth import SCOPES, authenticate, create_token
from app.models import (
BacktestPreview,
BacktestResult,
BacktestRun,
MCPAudit,
MCPToken,
Pnl,
now,
)
from tests.test_backtests import candidate, execute, setup
ENDPOINT = "/api/v1/mcp/"
@pytest.fixture
async def mcp_app(settings):
from app.main import create_app
from app.models import Base
from app.worldquant import WqClient
settings.mcp_enabled = True
def no_network(request):
raise AssertionError("No real platform")
application = create_app(settings, WqClient(settings, transport=httpx.MockTransport(no_network)))
async with application.state.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
ready, stop = asyncio.Event(), asyncio.Event()
async def lifespan():
async with application.router.lifespan_context(application):
ready.set()
await stop.wait()
task = asyncio.create_task(lifespan())
await ready.wait()
try:
await setup(application)
yield application
finally:
stop.set()
await task
async def credentials(app, scopes=SCOPES):
async with app.state.sessions.begin() as db:
row, secret = await create_token(db, "synthetic test", scopes)
principal = await authenticate(db, secret)
return principal, secret
def submission(key="batch-1", items=None, **extra):
return {"name": "MCP batch", "candidates": items or [candidate()], "idempotency_key": key, **extra}
async def invoke(app, principal, name, arguments=None):
result = await app.state.mcp.invoke(principal, name, arguments or {})
assert not result.is_error, result.structured_content
return result.structured_content
async def test_http_auth_and_discovery(mcp_app):
principal, secret = await credentials(mcp_app, {"research:read"})
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=mcp_app), base_url="http://testserver") as client:
payload = {"jsonrpc": "2.0", "id": 1, "method": "tools/list"}
assert (await client.post(ENDPOINT, json=payload)).status_code == 401
headers = {"Authorization": f"Bearer {secret}", "Accept": "application/json, text/event-stream"}
response = await client.post(ENDPOINT, json=payload, headers=headers)
assert response.status_code == 200, response.text
names = {t["name"] for t in response.json()["result"]["tools"]}
assert "search_backtests" in names and "submit_backtests" not in names
denied = {"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "submit_backtests", "arguments": submission()}}
assert (await client.post(ENDPOINT, json=denied, headers=headers)).status_code == 403
assert (await client.get("/api/v1/account", headers=headers)).status_code == 401
assert (await client.post(ENDPOINT, json=payload, headers=headers | {"Origin": "https://bad.test"})).status_code == 403
assert (await client.post(ENDPOINT, json=payload, headers=headers | {"Host": "bad.test"})).status_code == 403
async with mcp_app.state.sessions.begin() as db:
row = await db.get(MCPToken, principal.token_id)
row.revoked_at = now()
assert (await client.post(ENDPOINT, json=payload, headers=headers)).status_code == 401
async def test_submit_replay_duplicates_and_rotation(mcp_app):
principal, _ = await credentials(mcp_app)
result = await invoke(mcp_app, principal, "submit_backtests", submission())
second, _ = await credentials(mcp_app)
replay = await invoke(mcp_app, second, "submit_backtests", submission())
assert replay == result
conflict = await mcp_app.state.mcp.invoke(second, "submit_backtests", submission(items=[candidate(1)]))
assert conflict.is_error and conflict.structured_content["error"]["code"] == "IDEMPOTENCY_CONFLICT"
duplicate = await mcp_app.state.mcp.invoke(second, "submit_backtests", submission("different"))
assert duplicate.structured_content["error"]["code"] == "DUPLICATE_INPUT"
rerun = await invoke(mcp_app, second, "submit_backtests", submission("different", duplicate_policy="rerun"))
assert rerun["backtest_run_id"] != result["backtest_run_id"]
async with mcp_app.state.sessions() as db:
assert await db.scalar(select(func.count()).select_from(BacktestRun)) == 2
assert await db.scalar(select(func.count()).select_from(BacktestPreview)) == 2
audits = list(await db.scalars(select(MCPAudit)))
assert len(audits) == 5
assert result["source"]["kind"] == "mcp"
assert "#backtests?run_id=" in result["web_url"]
async def test_validation_and_within_batch_atomic(mcp_app):
principal, _ = await credentials(mcp_app)
incomplete = candidate()
incomplete["settings"] = {"region": "USA", "universe": "TOP3000", "delay": 1}
result = await mcp_app.state.mcp.invoke(principal, "submit_backtests", submission(items=[incomplete]))
assert result.structured_content["error"]["code"] == "INVALID_INPUT"
result = await mcp_app.state.mcp.invoke(principal, "submit_backtests", submission(items=[candidate(), candidate() | {"client_item_id": "second"}]))
assert result.structured_content["error"]["code"] == "DUPLICATE_INPUT"
# Failed requests did not consume the key or leave previews.
await invoke(mcp_app, principal, "submit_backtests", submission())
async with mcp_app.state.sessions() as db:
assert await db.scalar(select(func.count()).select_from(BacktestPreview)) == 1
async def test_results_control_and_evidence(mcp_app):
principal, _ = await credentials(mcp_app)
result = await invoke(mcp_app, principal, "submit_backtests", submission(items=[candidate(0), candidate(1)]))
rid = result["backtest_run_id"]
control = {"run_id": rid, "action": "pause", "expected_version": 1, "idempotency_key": "pause"}
paused = await invoke(mcp_app, principal, "control_backtest", control)
assert paused == await invoke(mcp_app, principal, "control_backtest", control)
await invoke(mcp_app, principal, "control_backtest", control | {"action": "resume", "expected_version": 2, "idempotency_key": "resume"})
await execute(mcp_app, mcp_app.state.runner.backtests, rid)
data = await invoke(mcp_app, principal, "get_backtest_results", {"run_id": rid, "limit": 1})
assert data["has_more"] and data["items"][0]["metrics"]["is"]["sharpe"] is None
item = data["items"][0]
async with mcp_app.state.sessions.begin() as db:
saved = await db.get(BacktestResult, item["id"])
saved.snapshot = {**saved.snapshot, "is": {"checks": [{"name": "a", "result": "FAIL"}, {"name": "b", "result": "NEW_STATUS"}]}}
data = await invoke(mcp_app, principal, "get_backtest_results", {"run_id": rid})
assert data["items"][0]["checks"]["counts"]["FAIL"] == 1
assert data["items"][0]["checks"]["counts"]["UNKNOWN"] == 1
pnl = await invoke(mcp_app, principal, "get_backtest_artifact", {"item_id": item["id"], "kind": "pnl"})
assert pnl["status"] == "not_cached"
snapshot = await invoke(mcp_app, principal, "get_backtest_artifact", {"item_id": item["id"], "kind": "snapshot", "limit": 1})
assert snapshot["has_more"]
history = await invoke(mcp_app, principal, "search_backtests", {"candidates": [candidate()], "source": "mcp"})
assert history["total"] == 1 and history["items"][0]["match_type"] == "exact_input"
progress = await invoke(mcp_app, principal, "get_backtest", {"run_id": rid, "after": 0, "event_limit": 1})
assert progress["events"]["has_more"] and progress["submission_counts"]["actual_platform_consumption"] is None
async def test_official_sdk_client_and_error_contract(mcp_app):
import httpx2
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
principal, secret = await credentials(mcp_app)
async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=mcp_app),
headers={"Authorization": f"Bearer {secret}"}) as http:
async with streamable_http_client("http://testserver/api/v1/mcp/", http_client=http) as streams:
async with ClientSession(streams[0], streams[1]) as client:
await client.initialize()
listed = await client.list_tools()
assert len(listed.tools) == 11
caps = await client.call_tool("get_research_capabilities", {})
assert caps.structured_content["max_candidates"] == 100
result = await client.call_tool("submit_backtests", submission())
assert not result.is_error, result
rid = result.structured_content["backtest_run_id"]
await execute(mcp_app, mcp_app.state.runner.backtests, rid)
results = await client.call_tool("get_backtest_results", {"run_id": rid})
assert results.structured_content["items"][0]["persistence_status"] == "saved"
bad = await client.call_tool("submit_backtests", submission("duplicate"))
assert bad.is_error and bad.structured_content["error"]["code"] == "DUPLICATE_INPUT"
await client.call_tool("get_backtest", {"run_id": rid})
async def test_expiry_binding_disabled_and_audit_redaction(app, mcp_app):
from fastapi import HTTPException
from app.models import Account
principal, secret = await credentials(mcp_app)
await mcp_app.state.mcp.invoke(principal, "get_research_capabilities", {}, request_id=secret)
async with mcp_app.state.sessions.begin() as db:
audit = await db.scalar(select(MCPAudit))
assert secret not in audit.request_id
row = await db.get(MCPToken, principal.token_id)
row.expires_at = now() - timedelta(seconds=1)
async with mcp_app.state.sessions() as db:
with pytest.raises(HTTPException):
await authenticate(db, secret)
_, another = await credentials(mcp_app)
async with mcp_app.state.sessions.begin() as db:
account = await db.get(Account, 1)
account.wq_user_id = "CHANGED"
async with mcp_app.state.sessions() as db:
with pytest.raises(HTTPException):
await authenticate(db, another)
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as client:
assert (await client.post(ENDPOINT, json={})).status_code == 404
async def test_metadata_refresh_and_pnl_pagination(mcp_app):
from app.worldquant import WqClient
from tests.research_metadata_fake import response
principal, _ = await credentials(mcp_app)
caps = await invoke(mcp_app, principal, "get_research_metadata", {"query": {"kind": "settings"}})
assert caps["status"] == "not_cached"
await mcp_app.state.runner.client.close()
from tests.backtest_fake import Platform
platform = Platform()
client = WqClient(mcp_app.state.settings, transport=httpx.MockTransport(lambda request: response(request) or platform(request)))
await client.authenticate("synthetic@example.com", "synthetic-password")
mcp_app.state.runner.client = client
mcp_app.state.runner.backtests.client = client
operators = await invoke(mcp_app, principal, "refresh_research_data", {"query": {"kind": "operators"}})
assert operators["status"] == "completed"
page = await invoke(mcp_app, principal, "get_research_metadata", {"query": {"kind": "operators", "limit": 1}})
assert page["total"] == 2 and page["has_more"]
await invoke(mcp_app, principal, "refresh_research_data", {"query": {"kind": "settings"}})
await invoke(mcp_app, principal, "refresh_research_data", {"query": {"kind": "field_availability", "scope": {"region": "USA", "universe": "TOP3000", "delay": 1}, "field_id": "close"}})
invalid = await mcp_app.state.mcp.invoke(principal, "submit_backtests", submission(items=[candidate(region="INVALID")]))
assert invalid.structured_content["error"]["code"] == "UNSUPPORTED_SETTINGS"
run = await invoke(mcp_app, principal, "submit_backtests", submission())
await execute(mcp_app, mcp_app.state.runner.backtests, run["backtest_run_id"])
result = await invoke(mcp_app, principal, "get_backtest_results", {"run_id": run["backtest_run_id"]})
item = result["items"][0]
async with mcp_app.state.sessions.begin() as db:
db.add(Pnl(alpha_id=item["alpha_id"], raw={}, points=[{"date": f"2025-01-0{i}", "value": i} for i in range(1, 4)]))
pnl = await invoke(mcp_app, principal, "get_backtest_artifact", {"item_id": item["id"], "kind": "pnl", "limit": 1, "date_from": "2025-01-02"})
assert pnl["total"] == 2 and pnl["items"][0]["value"] == 2 and pnl["has_more"]
async def test_lost_http_response_replays_without_new_run(mcp_app):
import json
_, secret = await credentials(mcp_app)
class LoseResponse(httpx.ASGITransport):
dropped = False
async def handle_async_request(self, request):
response = await super().handle_async_request(request)
if not self.dropped:
self.dropped = True
await response.aread()
raise httpx.ReadError("synthetic lost response", request=request)
return response
transport = LoseResponse(app=mcp_app)
async with httpx.AsyncClient(transport=transport, base_url="http://testserver",
headers={"Authorization": f"Bearer {secret}", "Accept": "application/json, text/event-stream"}) as client:
payload = {"jsonrpc": "2.0", "id": "lost", "method": "tools/call",
"params": {"name": "submit_backtests", "arguments": submission()}}
with pytest.raises(httpx.ReadError):
await client.post(ENDPOINT, json=payload)
response = await client.post(ENDPOINT, json=payload)
assert response.status_code == 200
assert not response.json()["result"]["isError"]
async with mcp_app.state.sessions() as db:
assert await db.scalar(select(func.count()).select_from(BacktestRun)) == 1
audits = list(await db.scalars(select(MCPAudit)))
assert all("wqmcp_" not in json.dumps(a.__dict__, default=str) for a in audits)
async def test_refresh_job_error_pages_and_kind_isolation(mcp_app):
from app.models import Job, JobItem
principal, _ = await credentials(mcp_app)
async with mcp_app.state.sessions.begin() as db:
db.add(Job(id="refresh", kind="pnl_refresh", failed=2))
db.add(Job(id="auth", kind="connect"))
await db.flush()
db.add_all([JobItem(job_id="refresh", alpha_id=str(i), error="synthetic error") for i in range(2)])
result = await invoke(mcp_app, principal, "get_refresh_job", {"job_id": "refresh", "limit": 1})
assert result["errors"]["total"] == 2 and result["errors"]["has_more"]
result = await invoke(mcp_app, principal, "get_refresh_job", {"job_id": "refresh", "offset": 1})
assert result["errors"]["items"][0]["alpha_id"] == "1"
error = await mcp_app.state.mcp.invoke(principal, "get_refresh_job", {"job_id": "auth"})
assert error.structured_content["error"]["code"] == "NOT_FOUND"
+78
View File
@@ -0,0 +1,78 @@
"""Browser PAT lifecycle and authentication boundaries, with no upstream calls."""
from datetime import timedelta
import httpx
import pytest
from fastapi import HTTPException
from sqlalchemy import select
from app.mcp_api.auth import authenticate
from app.models import Account, MCPToken, now
async def test_token_browser_lifecycle(app, logged_in):
client = logged_in
# Administration remains available when the MCP transport is disabled.
initial = (await client.get("/api/v1/mcp-tokens")).json()
assert initial["enabled"] is False and initial["can_create"] is False
assert (await client.post("/api/v1/mcp-tokens", json={"name": "reader"})).status_code == 422
async with app.state.sessions.begin() as db:
account = await db.get(Account, 1)
account.wq_user_id = "synthetic-user"
issued = await client.post("/api/v1/mcp-tokens", json={"name": " reader "})
assert issued.status_code == 201
assert issued.headers["cache-control"] == "no-store"
token = issued.json()
assert token["name"] == "reader" and token["scopes"] == ["research:read"]
assert token["status"] == "active" and token["token"].startswith("wqmcp_")
async with app.state.sessions() as db:
principal = await authenticate(db, token["token"])
row = await db.get(MCPToken, principal.token_id)
assert row.token_hash != token["token"]
assert 89 < (row.expires_at.replace(tzinfo=now().tzinfo) - now()).total_seconds() / 86400 <= 90
page = await client.get("/api/v1/mcp-tokens?limit=1")
assert token["token"] not in page.text and "token_hash" not in page.text
assert page.json()["items"][0]["id"] == token["id"]
second = await client.post("/api/v1/mcp-tokens", json={"name": "executor", "days": 1, "scopes": ["research:read", "backtests:execute"]})
assert second.status_code == 201
page = (await client.get("/api/v1/mcp-tokens?limit=1")).json()
assert page["total"] == 2 and page["has_more"]
assert (await client.get("/api/v1/mcp-tokens?limit=1&offset=1")).json()["items"][0]["id"] == token["id"]
revoked = await client.post(f'/api/v1/mcp-tokens/{token["id"]}/revoke')
again = await client.post(f'/api/v1/mcp-tokens/{token["id"]}/revoke')
assert revoked.json() == again.json() and revoked.json()["status"] == "revoked"
assert "token" not in revoked.json()
async with app.state.sessions() as db:
with pytest.raises(HTTPException) as exc:
await authenticate(db, token["token"])
assert exc.value.status_code == 401
assert (await client.post("/api/v1/mcp-tokens/missing/revoke")).status_code == 404
async def test_token_browser_security_and_validation(app, logged_in):
client = logged_in
async with app.state.sessions.begin() as db:
(await db.get(Account, 1)).wq_user_id = "synthetic-user"
for body in [
{"name": " "}, {"name": "x", "days": 0}, {"name": "x", "days": 366},
{"name": "x", "days": 1.5}, {"name": "x", "scopes": ["backtests:execute"]},
{"name": "x", "scopes": ["research:read", "admin"]}, {"name": "x", "account_id": 2},
]:
assert (await client.post("/api/v1/mcp-tokens", json=body)).status_code == 422
assert (await client.post("/api/v1/mcp-tokens", json={"name": "x"}, headers={"X-WQ-Request": ""})).status_code == 403
assert (await client.post("/api/v1/mcp-tokens", json={"name": "x"}, headers={"Origin": "https://evil.test"})).status_code == 403
token = (await client.post("/api/v1/mcp-tokens", json={"name": "x"})).json()
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://testserver") as outsider:
headers = {"Authorization": f'Bearer {token["token"]}', "X-WQ-Request": "1"}
assert (await outsider.get("/api/v1/mcp-tokens")).status_code == 401
assert (await outsider.get("/api/v1/mcp-tokens", headers=headers)).status_code == 401
assert (await outsider.post("/api/v1/mcp-tokens", json={"name": "escalation"}, headers=headers)).status_code == 401
assert (await outsider.post(f'/api/v1/mcp-tokens/{token["id"]}/revoke', headers=headers)).status_code == 401
async with app.state.sessions.begin() as db:
(await db.get(Account, 1)).wq_user_id = "different-user"
assert (await client.get("/api/v1/mcp-tokens")).json()["items"][0]["status"] == "invalid_binding"
async with app.state.sessions.begin() as db:
row = await db.scalar(select(MCPToken))
row.expires_at = now() - timedelta(days=1)
assert (await client.get("/api/v1/mcp-tokens")).json()["items"][0]["status"] == "expired"
+159
View File
@@ -105,6 +105,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/b5/f0/f2ed1de154e15b107dc692262395b3c17fc34eafe2a78fc2115931561730/asyncpg-0.31.0-cp312-cp312-win_amd64.whl", hash = "sha256:72d6bdcbc93d608a1158f17932de2321f68b1a967a13e014998db87a72ed3186", size = 597175, upload-time = "2025-11-24T23:26:02.564Z" },
]
[[package]]
name = "attrs"
version = "26.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/9a/8e/82a0fe20a541c03148528be8cac2408564a6c9a0cc7e9171802bc1d26985/attrs-26.1.0.tar.gz", hash = "sha256:d03ceb89cb322a8fd706d4fb91940737b6642aa36998fe130a9bc96c985eff32", size = 952055, upload-time = "2026-03-19T14:22:25.026Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
]
[[package]]
name = "certifi"
version = "2026.7.22"
@@ -424,6 +433,33 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/78/f7/18a1afcd64f35314b68c1f23afcd9994d0bc13e65cc77517afff4e83986d/jiter-0.16.0-graalpy312-graalpy250_312_native-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:64d613743df53199b1aa256a7d328340da6d7078aac7705a7db9d7a791e9cfd2", size = 343885, upload-time = "2026-06-29T13:05:12.087Z" },
]
[[package]]
name = "jsonschema"
version = "4.26.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "jsonschema-specifications" },
{ name = "referencing" },
{ name = "rpds-py" },
]
sdist = { url = "https://files.pythonhosted.org/packages/b3/fc/e067678238fa451312d4c62bf6e6cf5ec56375422aee02f9cb5f909b3047/jsonschema-4.26.0.tar.gz", hash = "sha256:0c26707e2efad8aa1bfc5b7ce170f3fccc2e4918ff85989ba9ffa9facb2be326", size = 366583, upload-time = "2026-01-07T13:41:07.246Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/69/90/f63fb5873511e014207a475e2bb4e8b2e570d655b00ac19a9a0ca0a385ee/jsonschema-4.26.0-py3-none-any.whl", hash = "sha256:d489f15263b8d200f8387e64b4c3a75f06629559fb73deb8fdfb525f2dab50ce", size = 90630, upload-time = "2026-01-07T13:41:05.306Z" },
]
[[package]]
name = "jsonschema-specifications"
version = "2025.9.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "referencing" },
]
sdist = { url = "https://files.pythonhosted.org/packages/19/74/a633ee74eb36c44aa6d1095e7cc5569bebf04342ee146178e2d36600708b/jsonschema_specifications-2025.9.1.tar.gz", hash = "sha256:b540987f239e745613c7a9176f3edb72b832a4ac465cf02712288397832b5e8d", size = 32855, upload-time = "2025-09-08T01:34:59.186Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/41/45/1a4ed80516f02155c51f51e8cedb3c1902296743db0bbc66608a0db2814f/jsonschema_specifications-2025.9.1-py3-none-any.whl", hash = "sha256:98802fee3a11ee76ecaca44429fda8a41bff98b00a0f2838151b113f210cc6fe", size = 18437, upload-time = "2025-09-08T01:34:57.871Z" },
]
[[package]]
name = "logfire-api"
version = "5.0.0"
@@ -464,6 +500,44 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/e5/f1/216fc1bbfd74011693a4fd837e7026152e89c4bcf3e77b6692fba9923123/markupsafe-3.0.3-cp312-cp312-win_arm64.whl", hash = "sha256:35add3b638a5d900e807944a078b51922212fb3dedb01633a8defc4b01a3c85f", size = 13906, upload-time = "2025-09-27T18:36:40.689Z" },
]
[[package]]
name = "mcp"
version = "2.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
{ name = "httpx2" },
{ name = "jsonschema" },
{ name = "mcp-types" },
{ name = "opentelemetry-api" },
{ name = "pydantic" },
{ name = "pyjwt", extra = ["crypto"] },
{ name = "python-multipart" },
{ name = "pywin32", marker = "sys_platform == 'win32'" },
{ name = "sse-starlette" },
{ name = "starlette" },
{ name = "typing-extensions" },
{ name = "typing-inspection" },
{ name = "uvicorn", marker = "sys_platform != 'emscripten'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/76/31/ac54fb0fdd5b37de704486e288bba4fbbb463f24cfcfedbede407b854513/mcp-2.2.0.tar.gz", hash = "sha256:2dc37ecb1974becdcebdbf7561e7c15a07dbbf20ba21ba16c3593b3038b3afbd", size = 4084129, upload-time = "2026-09-07T16:06:23.439Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/1b/ff/8e7eade68b8a28f7da0ed1085544341b51f9c935dbf6b95c76b7edfea6a0/mcp-2.2.0-py3-none-any.whl", hash = "sha256:bde982589473a060ae145e3406e9a5333fe538c97229ba841f5a7f92be004f81", size = 365656, upload-time = "2026-09-07T16:06:19.711Z" },
]
[[package]]
name = "mcp-types"
version = "2.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pydantic" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/ae/91/762d7755d971aff8a28d75f7961656148edf27875c8026e6385aaab08ae7/mcp_types-2.2.0.tar.gz", hash = "sha256:d3ed53703ddd10d9c6399f29d322bb66f3f67ab41348ac8556ba23e07fedefad", size = 65892, upload-time = "2026-09-07T16:06:25.187Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/8f/d7/6ffba5d8cd5dd9b8a19478875c50e04945314ba5074e84d749283f27f62d/mcp_types-2.2.0-py3-none-any.whl", hash = "sha256:ea476b73ee86709ab5abc9452385ed36cc05907e582355622e294595c9a04f13", size = 69106, upload-time = "2026-09-07T16:06:21.461Z" },
]
[[package]]
name = "openai"
version = "3.8.0"
@@ -627,6 +701,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/71/46/17f022dd3e953bf20a04a028a21ec746d942f8d2af30fa0f124fa0e6a684/pygments-2.21.0-py3-none-any.whl", hash = "sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9", size = 1250147, upload-time = "2026-08-17T08:02:44.912Z" },
]
[[package]]
name = "pyjwt"
version = "2.13.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/3b/81/58d0ac84e1ef3a3843791d6954d94c0b33d526c75eeb1efbce9d0a4c4077/pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423", size = 107515, upload-time = "2026-05-21T19:54:36.618Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" },
]
[package.optional-dependencies]
crypto = [
{ name = "cryptography" },
]
[[package]]
name = "pytest"
version = "9.1.1"
@@ -665,6 +753,25 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/0d/17/c5c6b53ddc18f297992099b3d9ec16c855c0ccc83263a21fe4d1c625ec6c/python_dotenv-1.2.3-py3-none-any.whl", hash = "sha256:904552145e8bfed22162c09dab1c2b9b54fefa7b23ba780f4f26ca0316b0f0d9", size = 22780, upload-time = "2026-08-16T16:54:52.473Z" },
]
[[package]]
name = "python-multipart"
version = "0.0.32"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/5b/42/55c32bb9b12693c092ad250a0e82edb5b31ddeda6eb772de5f308b3804ad/python_multipart-0.0.32.tar.gz", hash = "sha256:be54b7f3fa167bb83e4fcd936b887b708f4e57fe75911c02aebf53efaf8d938e", size = 46881, upload-time = "2026-06-04T16:18:58.647Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e1/04/e8135ebd1ad02c56ec633277529b2602ff99ff634be76cdba5744cf554fd/python_multipart-0.0.32-py3-none-any.whl", hash = "sha256:ff6d3f776f16878c894e52e107296ffc890e913c611b1a4ec6c44e2821fe2e23", size = 30042, upload-time = "2026-06-04T16:18:57.319Z" },
]
[[package]]
name = "pywin32"
version = "312"
source = { registry = "https://pypi.org/simple" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/83/ff/32aa7d2ed0ab12b323aaa64f9b75e6ad4f8fd09f9ccfc28c79414d46838d/pywin32-312-cp312-cp312-win32.whl", hash = "sha256:dab4f65ac9c4e48400a2a0530c46c3c579cd5905ecd11b80692373915269208b", size = 6371877, upload-time = "2026-06-04T07:49:28.836Z" },
{ url = "https://files.pythonhosted.org/packages/03/d9/77040d3b43df3f3be32ea289433d660d2727f5ba327bc73be835127d9d60/pywin32-312-cp312-cp312-win_amd64.whl", hash = "sha256:b457f6d628a47e8a7346ce22acb7e1a46a4a78b52e1d17e1af56871bd19a93bc", size = 6914841, upload-time = "2026-06-04T07:49:31.85Z" },
{ url = "https://files.pythonhosted.org/packages/e3/cc/7b1ec671775756020a0ee7f4feeaf3c568f0ab86bd3900088cf986937a92/pywin32-312-cp312-cp312-win_arm64.whl", hash = "sha256:6017c58e12f6809fbb0555b75df144c2922a9ffd18e4b9b5afa863b6c1a9d950", size = 6727901, upload-time = "2026-06-04T07:49:34.244Z" },
]
[[package]]
name = "pyyaml"
version = "6.0.3"
@@ -683,6 +790,20 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/1a/08/67bd04656199bbb51dbed1439b7f27601dfb576fb864099c7ef0c3e55531/pyyaml-6.0.3-cp312-cp312-win_arm64.whl", hash = "sha256:64386e5e707d03a7e172c0701abfb7e10f0fb753ee1d773128192742712a98fd", size = 140344, upload-time = "2025-09-25T21:32:22.617Z" },
]
[[package]]
name = "referencing"
version = "0.37.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "rpds-py" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/22/f5/df4e9027acead3ecc63e50fe1e36aca1523e1719559c499951bb4b53188f/referencing-0.37.0.tar.gz", hash = "sha256:44aefc3142c5b842538163acb373e24cce6632bd54bdb01b21ad5863489f50d8", size = 78036, upload-time = "2025-10-13T15:30:48.871Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2c/58/ca301544e1fa93ed4f80d724bf5b194f6e4b945841c5bfd555878eea9fcb/referencing-0.37.0-py3-none-any.whl", hash = "sha256:381329a9f99628c9069361716891d34ad94af76e461dcb0335825aecc7692231", size = 26766, upload-time = "2025-10-13T15:30:47.625Z" },
]
[[package]]
name = "regex"
version = "2026.9.3"
@@ -722,6 +843,29 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/a0/f4/c67b0b3f1b9245e8d266f0f112c500d50e5b4e83cb6f3b71b6528104182a/requests-2.34.2-py3-none-any.whl", hash = "sha256:2a0d60c172f83ac6ab31e4554906c0f3b3588d37b5cb939b1c061f4907e278e0", size = 73075, upload-time = "2026-05-14T19:25:26.443Z" },
]
[[package]]
name = "rpds-py"
version = "2026.6.3"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/aa/2a/9618a122aeb2a169a28b03889a2995fe297588964333d4a7d67bdf46e147/rpds_py-2026.6.3.tar.gz", hash = "sha256:1cebd1337c242e4ec2293e541f712b2da849b29f48f0c293684b71c0632625d4", size = 64051, upload-time = "2026-06-30T07:17:53.009Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5c/be/2e8974163072e7bab7df1a5acd54c4498e75e35d6d18b864d3a9d5dadc92/rpds_py-2026.6.3-cp312-cp312-macosx_10_12_x86_64.whl", hash = "sha256:a0811d33247c3d6128a3001d763f2aa056bb3425204335400ac54f89eec3a0d0", size = 343691, upload-time = "2026-06-30T07:15:14.96Z" },
{ url = "https://files.pythonhosted.org/packages/a4/73/319dfa745dd668efe89309141ded489126461fcecd2b8f3a3cda185129b6/rpds_py-2026.6.3-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:538949e262e46caa31ac01bdb3c1e8f642622922cacbabbae6a8445d9dc33eaf", size = 338542, upload-time = "2026-06-30T07:15:16.267Z" },
{ url = "https://files.pythonhosted.org/packages/21/63/4239893be1c4d09b709b1a8f6be4188f0870084ff547f46606b8a75f1b03/rpds_py-2026.6.3-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:55927d532399c2c646100ff7feb48eaa940ad70f42cd68e1328f3ded9f81ca24", size = 368180, upload-time = "2026-06-30T07:15:17.62Z" },
{ url = "https://files.pythonhosted.org/packages/1c/ca/9c5de382225234ceb37b1844ebdb140db12b2a278bb9efe2fcd19f6c82ce/rpds_py-2026.6.3-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:f56f1695bc5c0871cbc33dc0130fcf503aab0c57dcc5a6700a4f49eba4f2652e", size = 375067, upload-time = "2026-06-30T07:15:18.952Z" },
{ url = "https://files.pythonhosted.org/packages/87/dc/863f69d1bf04ade34b7fe0d59b9fdf6f0135fe2d7cbca74f1d665589559d/rpds_py-2026.6.3-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:270b293dae9058fc9fcedab50f13cebf46fb8ed1d1d54e0521a9da5d6b211975", size = 490509, upload-time = "2026-06-30T07:15:20.434Z" },
{ url = "https://files.pythonhosted.org/packages/ce/ef/eac16a12048b45ec7c7fa94f2be3438a5f26bf9cc8580b18a1cfd609b7f6/rpds_py-2026.6.3-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:127565fead0a10943b282957bd5447804ff3160ad79f2ad2635e6d249e380680", size = 382754, upload-time = "2026-06-30T07:15:21.831Z" },
{ url = "https://files.pythonhosted.org/packages/04/8f/d2f3f532616be4d06c316ef119683e832bd3d41e112bf3a88f4151c95b17/rpds_py-2026.6.3-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:ecabd69db66de867690f9797f2f8fa27ba501bbc24540cbdbdc649cd15888ba6", size = 366189, upload-time = "2026-06-30T07:15:23.371Z" },
{ url = "https://files.pythonhosted.org/packages/e3/29/41a7b0e98a4b44cd676ab7598419623373eb43b20be68c084935c1a8cf88/rpds_py-2026.6.3-cp312-cp312-manylinux_2_31_riscv64.whl", hash = "sha256:58eadac9cd119677b60e1cf8ac4052f35949d71b8a9e5556efccbe82533cf22a", size = 377750, upload-time = "2026-06-30T07:15:24.659Z" },
{ url = "https://files.pythonhosted.org/packages/2e/05/ecda0bec46f9a1565090bcdc941d023f6a25aff85fda28f89f8d19878152/rpds_py-2026.6.3-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:7491ee23305ac3eb59e492b6945881f5cd77a6f731061a3f25b77fd40f9e99a4", size = 395576, upload-time = "2026-06-30T07:15:25.987Z" },
{ url = "https://files.pythonhosted.org/packages/68/a8/6ed52f03ee6cb854ce78785cc9a9a672eb880e83fd7224d471f667d151f1/rpds_py-2026.6.3-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:2c99f7e8ccb3dd6e3e4bfeac657a7b208c9bac8075f4b078c02d7404c34107fa", size = 543807, upload-time = "2026-06-30T07:15:27.356Z" },
{ url = "https://files.pythonhosted.org/packages/8f/d6/156c0d3eea27ba09b92562ba2364ba124c0a061b199e17eac637cd25a5e2/rpds_py-2026.6.3-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:62698275682bf121181861295c9181e789030a2d516071f5b8f3c23c170cd0fc", size = 611187, upload-time = "2026-06-30T07:15:28.931Z" },
{ url = "https://files.pythonhosted.org/packages/f1/31/774212ed989c62f7f310220089f9b0a3fb8f40f5443d1727abd5d9f52bc9/rpds_py-2026.6.3-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:a214c993455f99a89aaeadc9b21241900037adc9d97203e374d75513c5911822", size = 573030, upload-time = "2026-06-30T07:15:30.553Z" },
{ url = "https://files.pythonhosted.org/packages/c9/50/22f73127a41f1ce4f87fe39aadfb9a126345801c274aa93ae88456249327/rpds_py-2026.6.3-cp312-cp312-win32.whl", hash = "sha256:501f9f04a588d6a09179368c57071301445191767c64e4b52a6aa9871f1ef5ed", size = 202185, upload-time = "2026-06-30T07:15:32.027Z" },
{ url = "https://files.pythonhosted.org/packages/04/3a/f0ee4d4dde9d3b69dedf1b5f74e7a40017046d55052d173e418c6a94f960/rpds_py-2026.6.3-cp312-cp312-win_amd64.whl", hash = "sha256:2c958bf94822e9290a40aaf2a822d4bc5c88099093e3948ad6c571eca9272e5f", size = 220394, upload-time = "2026-06-30T07:15:33.359Z" },
{ url = "https://files.pythonhosted.org/packages/f3/83/3382fe37f809b59f02aac04dbc4e765b480b46ee0227ed516e3bdc4d3dfc/rpds_py-2026.6.3-cp312-cp312-win_arm64.whl", hash = "sha256:22bffe6042b9bcb0822bcd1955ec00e245daf17b4344e4ed8e9551b976b63e96", size = 215753, upload-time = "2026-06-30T07:15:34.778Z" },
]
[[package]]
name = "ruff"
version = "0.16.6"
@@ -781,6 +925,19 @@ asyncio = [
{ name = "greenlet" },
]
[[package]]
name = "sse-starlette"
version = "3.4.11"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
{ name = "starlette" },
]
sdist = { url = "https://files.pythonhosted.org/packages/2b/54/6767bb789b2f2fed6e0f953df949cd39dc263a384c1b65a95232598621d6/sse_starlette-3.4.11.tar.gz", hash = "sha256:1bae716c02f3e6f294be41ff333220692dae7c3cbab077c900f159676719dade", size = 34972, upload-time = "2026-09-05T12:11:04.607Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/98/6a/2ba3ed4a69babf3afdddf7d8314a48d87562c0a442206bbc2a1b50d5efc0/sse_starlette-3.4.11-py3-none-any.whl", hash = "sha256:c7b2244bdff016fe7f64e10075e89a3e6bbf899649cc89b0fe884b5545042453", size = 17122, upload-time = "2026-09-05T12:11:03.195Z" },
]
[[package]]
name = "starlette"
version = "1.6.0"
@@ -954,6 +1111,7 @@ dependencies = [
{ name = "cryptography" },
{ name = "fastapi" },
{ name = "httpx" },
{ name = "mcp" },
{ name = "pydantic-ai-slim", extra = ["openai"] },
{ name = "pydantic-settings" },
{ name = "sqlalchemy", extra = ["asyncio"] },
@@ -976,6 +1134,7 @@ requires-dist = [
{ name = "cryptography", specifier = ">=44,<50" },
{ name = "fastapi", specifier = ">=0.115,<1" },
{ name = "httpx", specifier = ">=0.28,<1" },
{ name = "mcp", specifier = ">=2,<3" },
{ name = "pydantic-ai-slim", extras = ["openai"], specifier = "==1.97.0" },
{ name = "pydantic-settings", specifier = ">=2.8,<3" },
{ name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.38,<2.1" },
+1
View File
@@ -15,6 +15,7 @@ x-backend: &backend
WQ_EMAIL: ${WQ_EMAIL:?Set the Gitea WQ_EMAIL secret}
WQ_PASSWORD: ${WQ_PASSWORD:?Set the Gitea WQ_PASSWORD secret}
COOKIE_SECURE: "true"
MCP_ENABLED: ${MCP_ENABLED:-false}
AI_REQUEST_LIMIT: ${AI_REQUEST_LIMIT:-12}
AI_TOOL_LIMIT: ${AI_TOOL_LIMIT:-12}
AI_OUTPUT_TOKENS: ${AI_OUTPUT_TOKENS:-4096}
+1
View File
@@ -25,6 +25,7 @@ services:
ADMIN_PASSWORD: ${ADMIN_PASSWORD:?required}
ENCRYPTION_KEY: ${ENCRYPTION_KEY:?required}
PUBLIC_ORIGIN: https://${DOMAIN:?Set DOMAIN to your real hostname}
MCP_ENABLED: ${MCP_ENABLED:-false}
AI_REQUEST_LIMIT: ${AI_REQUEST_LIMIT:-12}
AI_TOOL_LIMIT: ${AI_TOOL_LIMIT:-12}
AI_OUTPUT_TOKENS: ${AI_OUTPUT_TOKENS:-4096}
+1
View File
@@ -24,6 +24,7 @@ services:
ADMIN_PASSWORD: ${ADMIN_PASSWORD:?required}
ENCRYPTION_KEY: ${ENCRYPTION_KEY:?required}
PUBLIC_ORIGIN: http://localhost:${LOCAL_PORT:-8080}
MCP_ENABLED: ${MCP_ENABLED:-false}
AI_REQUEST_LIMIT: ${AI_REQUEST_LIMIT:-12}
AI_TOOL_LIMIT: ${AI_TOOL_LIMIT:-12}
AI_OUTPUT_TOKENS: ${AI_OUTPUT_TOKENS:-4096}
+1
View File
@@ -16,6 +16,7 @@
| Variables(必填) | `DATABASE_NETWORK` | PostgreSQL 所在的现有 Docker 网络,例如 `1panel-network` |
| Variables(必填) | `PUBLIC_ORIGIN` | 实际 HTTPS 来源,例如 `https://alpha.your-domain.com`,不带路径或末尾 `/` |
| Variables(可选) | `ADMIN_USERNAME` | 初始管理员账号,默认 `admin` |
| Variables(可选) | `MCP_ENABLED` | 设置 `true` 启用 MCP;未配置时默认关闭 |
端口与 AI 限制使用 `compose.production.yaml` 的默认值,不需要在 Gitea 配置:`WEB_PORT=8112`、`AI_REQUEST_LIMIT=12`、`AI_TOOL_LIMIT=12`、`AI_OUTPUT_TOKENS=4096`、`AI_TIMEOUT=180`。需要调整时修改 Compose 中对应默认值;工作流不再读取这些同名 Gitea Variables。
+120
View File
@@ -0,0 +1,120 @@
# MCP 研究接入
MCP 让外部助手直接查询数据元信息、查回测历史、提交固定候选并读取结果。无需先建立特征、模板、变体或 QuantFlow。执行仍由网页共用的持久队列负责,回测页保存同一个运行。
## 启用与令牌
应用依赖官方 `mcp==2.2.0`,锁文件固定版本。增量迁移 `0010` 增加 `mcp_tokens`、`research_requests` 和 `mcp_audits`,不重写旧业务记录。
默认 `MCP_ENABLED=false`,入口返回 404。需要启用时,在目标部署配置中显式设置为 true,按该环境既有升级流程迁移并重建后端。生产仍限一个后端 worker。入口为 `PUBLIC_ORIGIN/api/v1/mcp/`;无尾斜杠地址会重定向。公网须使用 HTTPS,代理保留原始 Host。现有 Caddy 已保持 Host;Vite 开发代理也显式保持。
本功能不实现 OAuth;客户端必须能够自行配置 `Authorization: Bearer <PAT>`。PAT 不能作为 Cookie,也不能访问网页管理接口。无需添加 `X-WQ-Request`。若请求包含 Origin,须与 PUBLIC_ORIGIN 一致。
先在网页连接并确认 WorldQuant 账户身份。进入侧栏 **系统管理 → MCP Key**,填写名称、有效期并选择权限,即可创建 Key。默认只读、90 天;创建后明文仅显示一次,关闭、刷新或离开页面后无法找回。列表支持分页查看权限、到期时间和状态,并可撤销 Key。页面同时显示连接地址和 MCP 服务是否启用;管理页面不会自动打开服务开关。
管理接口沿用管理员 Cookie 会话和网页请求保护,MCP PAT 不能创建、查看或撤销 Key。明文不写入浏览器存储,也不传给网页研究助手。
也可在所需环境执行管理命令。以下是命令模板,不会由文档自动执行:
```bash
# 默认只读,90 天;--days 支持 1–365。
python -m app.cli mcp-token-create --name research-reader
# 需要对应业务权限时显式列出;必须包含 research:read。
python -m app.cli mcp-token-create --name research-executor \
--scope research:read --scope research:refresh \
--scope backtests:execute --scope backtests:control
python -m app.cli mcp-token-list
python -m app.cli mcp-token-revoke TOKEN_ID
```
容器中使用相应 Compose 配置的 `exec backend` 执行。签发成功后明文只输出一次,保管在客户端凭据存储中,不放入聊天、仓库或请求 ID。数据库只保存令牌哈希及绑定信息。每次 HTTP 请求重新验证过期、撤销、管理员和平台账户绑定。密码重置同时撤销 PAT;仅撤销 PAT 不会取消已受理的回测。
| 权限 | 可调用能力 |
| --- | --- |
| research:read | 能力、数据目录、元数据、历史、运行、结果、证据、刷新任务查询 |
| research:refresh | 显式更新元数据及 PnL 缓存;同时要求 read |
| backtests:execute | 直接提交固定候选;同时要求 read |
| backtests:control | 暂停、继续、停止、恢复采集;同时要求 read |
`tools/list` 只显示当前令牌可用的工具,实际调用仍再次验证权限。缺失/失效令牌返回 401,缺权限返回 403。
## 工具输入与返回
工具 schema 由 `tools/list` 提供,未知参数拒绝。所有业务返回为 `structuredContent`,并保留等价文本 JSON。业务失败置 `isError=true`,error 包含 code、message、retryable、retry_after、affected_items;HTTP 认证错误不伪装为正常工具结果。
| 工具 | 输入要点 |
| --- | --- |
| 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 |
| refresh_research_data | `{query:{kind,...}}`;kind 为 catalog/operators/settings/field_availability/pnl |
| get_refresh_job | `{job_id,limit?,offset?}`;错误列表独立分页 |
| search_backtests | 来源、reference、status、带时区起止时间、scope、q、候选精确匹配及分页 |
| submit_backtests | `{name,candidates,idempotency_key,duplicate_policy?,source?}` |
| get_backtest | `{run_id,after?,event_limit?}`,after 为事件游标 |
| get_backtest_results | `{run_id,item_ids?,limit?,offset?}` |
| get_backtest_artifact | `{item_id,kind,limit?,offset?,date_from?,date_to?}`,kind 为 snapshot/pnl |
| control_backtest | `{run_id,action,expected_version,idempotency_key}` |
metadata 的 operators 支持 q/category 和分页;settings 支持分页;field_availability 要求 field_id 和 scope。refresh 的 catalog 要求 scope,可选 dataset_id;pnl 要求 alpha_ids;availability 与读取使用相同范围字段。目录和 PnL 刷新返回 job_id,查询不会隐式刷新;另外三种刷新最多等待 30 秒,成功只返回快照引用,完整内容用读取工具获取。失败不发布半成品。
列表默认 25 项、最多 100 项;返回 total、offset、limit、has_more。快照证据按顶层 key/value 分页,嵌套内容完整保留;PnL 按日期过滤和分页。元数据仅提供字段与算子资料,不提供原始财务时间序列。
## 一轮研究示例
先读取能力和设置快照,发现字段并检查历史。用户授权本批执行后,提交以下形态的固定输入;设置仅为结构示例,实际范围需依据平台选项选择:
```json
{
"name": "价格排序基线",
"idempotency_key": "research-round-001",
"duplicate_policy": "reject",
"source": {"reference": "conversation-reference", "hypothesis": "基线对照"},
"candidates": [{
"client_item_id": "baseline",
"expression": "rank(close)",
"alpha_type": "REGULAR",
"settings": {
"instrumentType": "EQUITY", "region": "USA", "universe": "TOP3000", "delay": 1,
"decay": 0, "neutralization": "INDUSTRY", "truncation": 0.08,
"pasteurization": "ON", "unitHandling": "VERIFY", "nanHandling": "OFF",
"language": "FASTEXPR", "visualization": false, "maxTrade": "OFF", "maxPosition": "OFF"
}
}]
}
```
单次 1–100 项,client_item_id 唯一,每项设置完整,不做批次级参数合并。输入结构及已缓存设置组合参与校验;缺设置缓存时返回 validation=unknown,不宣称验证了平台组合、表达式语义或字段可用性。可先显式刷新设置消除这类未知。
提交自动保存固定预览和运行,返回 backtest_run_id、输入摘要、数量和 web_url。打开 web_url 可定位网页详情;来源显示“MCP 研究”,可以按来源筛选。后续 get_backtest 和 get_backtest_results 读取进度与结果。PnL 未缓存时返回 not_cached,通过显式刷新再读取。下一批可设置 source.parent_run_id 关联前一批。
## 幂等、重复与状态
- 成功请求按账户、操作和幂等键固定首次响应。相同请求重试返回原响应,不重新提交;换令牌不改变该范围。同键不同内容返回 IDEMPOTENCY_CONFLICT。失败校验不占用幂等键。
- 幂等判断先于重复查询。规范化摘要包含完整候选、顺序、设置、名称、来源说明和重复策略,不包含链路请求 ID 和令牌。
- 重复按平台完整输入指纹匹配,不判断数学等价。默认 reject 整批拒绝;返回历史记录及其状态,历史失败或跳过也属于输入匹配。明确 rerun 才创建新运行,不自动复用或跳过。匹配记录超页时通过 search_backtests 继续读取。
- 返回 ID 表示已持久化受理,不表示平台完成。执行继续使用原来的限流、退避与重启恢复;客户端断开不取消已受理运行。
- 保留平台、采集和持久化三层状态。completed 不等于检查 PASS。检查包含原始非通过内容及未知状态;指标缺失为 null。
- 历史指标来自当次固定快照,后续 Alpha 同步不改写。PnL 是另行采集的缓存,必须同时阅读 fetched_at,不能视为当次回测同时抓取。
- pause 阻止后续新提交;stop 跳过未提交项,已提交模拟继续采集,不能远程取消。停止后的剩余项须明确新建重跑。
- recover 只根据已有回执恢复采集,不重发未知模拟。无回执的 submission_unknown 仍需在网页核对原模拟。
- resume 保留解除账户级无期限调度阻塞的行为,响应 impact 明确是否解除;不会清除 Retry-After 截止时间。
- 控制使用 expected_version 和独立幂等键;成功重试不重复追加控制事件。统计区分候选、尝试、POST 请求和确认/未知接受数量,实际平台额度消耗保持未知。
审计只记录身份 ID、输入摘要、哈希后的链路请求标识、业务引用、结果码和耗时。关闭 MCP 开关不删除记录,也不停止已受理队列。备份需包含新增表;恢复后旧成功幂等键仍有效。
## 验证和限制
本地测试只使用模拟平台。后端 `uv run pytest -q tests/test_mcp.py` 包含官方 MCP ClientSession 的 HTTP 工具往返;浏览器测试在临时数据库里验证实际 MCP 提交的运行链接和来源。
独立 PostgreSQL 检查只接受本机 `wq_mcp_test` 数据库:
```bash
# backend 目录;变量必须指向专用、可丢弃的测试库。
MCP_TEST_DATABASE_URL=postgresql+asyncpg://USER:PASSWORD@127.0.0.1:PORT/wq_mcp_test \
uv run python -m tests.mcp_postgres
```
该脚本执行迁移、并发提交/控制、重启重放、回退及重升级,不用于个人库或生产库。生产启用、真实平台兼容性、真实额度和客户端实际凭据配置仍需另行授权验证。本功能不会恢复任何定时研究。
+3
View File
@@ -21,6 +21,7 @@ import { AppSidebar } from "./components/AppSidebar";
import "./components/sidebar.css";
import { api, patch, post } from "./api";
import type { Account, Job } from "./types";
import { MCPKeysPage } from "./pages/MCPKeysPage";
import { AccountPage } from "./pages/AccountPage";
import { DatasetPage } from "./pages/DatasetPage";
import { AlphaPage } from "./pages/AlphaPage";
@@ -318,6 +319,7 @@ export default function App() {
<main
className={`page-content ${page !== "account" ? "bounded-page" : "account-page"}`}
>
{page === "mcp-keys" && <MCPKeysPage />}
{pollError && (
<Banner
type="warning"
@@ -502,6 +504,7 @@ export default function App() {
datasets: datasetContext,
backtests: backtestContext,
account: { page: "account" as const },
"mcp-keys": { page: "account" as const },
}[page]
}
timezone={account?.timezone}
+7 -3
View File
@@ -1,7 +1,10 @@
import type { PageContext, UIAction } from "./types";
export type WorkspacePage = PageContext["page"];
const contextLabels: Record<WorkspacePage, (context: PageContext) => string> = {
export type WorkspacePage = PageContext["page"] | "mcp-keys";
const contextLabels: Record<
PageContext["page"],
(context: PageContext) => string
> = {
alphas: (context) =>
`上下文:${context.alpha_id ? `Alpha ${context.alpha_id}` : "Alpha 列表"}${context.selected_ids?.length ? ` · 已选 ${context.selected_ids.length} 条` : ""}`,
operators: () => "上下文:算子库",
@@ -24,8 +27,9 @@ export function contextLabel(context: PageContext): string {
}
export function pageFromHash(hash: string): WorkspacePage {
const value = hash.replace(/^#/, "");
const value = hash.replace(/^#/, "").split("?")[0];
const page = value.startsWith("quantflow/") ? "quantflow" : value;
if (page === "mcp-keys") return page;
return Object.hasOwn(contextLabels, page)
? (page as WorkspacePage)
: "alphas";
+11
View File
@@ -177,6 +177,17 @@ export function BacktestPage({
}
}, [action]);
useEffect(() => {
const locate = () => {
if (!location.hash.startsWith("#backtests?")) return;
const id = new URLSearchParams(location.hash.split("?")[1]).get("run_id");
if (id && /^[a-f0-9-]{36}$/i.test(id)) openRun(id);
};
locate();
window.addEventListener("hashchange", locate);
return () => window.removeEventListener("hashchange", locate);
}, []);
async function perform(work: () => Promise<void>) {
setBusy(true);
setError("");
+5 -2
View File
@@ -30,6 +30,7 @@ const navigation = [
{ id: "pipeline", label: "研究流水线", icon: IconBeaker, group: "研究编排" },
{ id: "quantflow", label: "QuantFlow", icon: IconBeaker, group: "研究编排" },
{ id: "alphas", label: "Alpha 管理", icon: IconGridView, group: "研究成果" },
{ id: "mcp-keys", label: "MCP Key", icon: IconCommand, group: "系统管理" },
{ id: "account", label: "个人信息", icon: IconUser, group: "" },
] as const;
@@ -172,7 +173,8 @@ export function AppSidebar({
</div>
)}
<nav className="sidebar-nav" aria-label="主导航">
{["数据与素材", "研究实验", "研究编排", "研究成果"].map((group) => (
{["数据与素材", "研究实验", "研究编排", "研究成果", "系统管理"].map(
(group) => (
<div key={group}>
{!collapsed && (
<div className="sidebar-group-heading">{group}</div>
@@ -198,7 +200,8 @@ export function AppSidebar({
</Tooltip>
))}
</div>
))}
),
)}
</nav>
<div className="sidebar-footer">
{!collapsed && (
+332
View File
@@ -0,0 +1,332 @@
import { useEffect, useState } from "react";
import { Banner, Button, Input, Toast } from "@douyinfe/semi-ui-19";
import { api, formatTime, post } from "../api";
import "./mcp-keys.css";
type Token = {
id: string;
name: string;
scopes: string[];
created_at: string;
expires_at: string;
revoked_at: string | null;
status: "active" | "expired" | "revoked" | "invalid_binding";
};
type TokenPage = {
items: Token[];
total: number;
has_more: boolean;
enabled: boolean;
endpoint: string;
can_create: boolean;
};
const scopes = [
["research:read", "读取研究数据", "查询目录、历史、运行与结果"],
["research:refresh", "刷新研究数据", "更新元数据和 PnL 缓存"],
["backtests:execute", "执行回测", "提交新的回测批次"],
["backtests:control", "控制回测", "暂停、继续、停止与恢复采集"],
] as const;
const labels = {
active: "有效",
expired: "已过期",
revoked: "已撤销",
invalid_binding: "账户绑定失效",
};
export function MCPKeysPage() {
const [data, setData] = useState<TokenPage | null>(null);
const [offset, setOffset] = useState(0);
const [version, setVersion] = useState(0);
const [loading, setLoading] = useState(true);
const [error, setError] = useState("");
const [name, setName] = useState("");
const [days, setDays] = useState("90");
const [permissions, setPermissions] = useState<string[]>(["research:read"]);
const [busy, setBusy] = useState(false);
const [created, setCreated] = useState<(Token & { token: string }) | null>(
null,
);
const [revoking, setRevoking] = useState<string | null>(null);
useEffect(() => {
let active = true;
setLoading(true);
api<TokenPage>(`/mcp-tokens?offset=${offset}&limit=25`)
.then((next) => {
if (active) {
setData(next);
setError("");
}
})
.catch((e: Error) => {
if (active) setError(e.message);
})
.finally(() => {
if (active) setLoading(false);
});
return () => {
active = false;
};
}, [offset, version]);
async function create(event: React.FormEvent) {
event.preventDefault();
if (busy) return;
setBusy(true);
setError("");
try {
const result = await post<Token & { token: string }>("/mcp-tokens", {
name: name.trim(),
days: Number(days),
scopes: permissions,
});
setCreated(result);
setName("");
setOffset(0);
setVersion((v) => v + 1);
} catch (e) {
setError((e as Error).message);
} finally {
setBusy(false);
}
}
async function revoke(id: string) {
setBusy(true);
try {
await post(`/mcp-tokens/${id}/revoke`);
if (created?.id === id) setCreated(null);
setRevoking(null);
setVersion((v) => v + 1);
Toast.success("MCP Key 已撤销");
} catch (e) {
setError((e as Error).message);
} finally {
setBusy(false);
}
}
async function copy(value: string) {
try {
await navigator.clipboard.writeText(value);
Toast.success("已复制");
} catch {
Toast.error("复制失败,请手动选择并复制");
}
}
return (
<section className="mcp-keys-page">
<header>
<div>
<h1>MCP Key</h1>
<p>管理 Codex 等客户端访问研究数据和执行回测的权限。</p>
</div>
<Button
disabled={loading || busy}
onClick={() => setVersion((v) => v + 1)}
>
刷新列表
</Button>
</header>
{error && <Banner type="danger" description={error} />}
{data && !data.enabled && (
<Banner
type="warning"
description="MCP 服务未启用。可以管理 Key,启用服务后客户端才能连接。"
/>
)}
{data && !data.can_create && (
<Banner
type="warning"
description="请先在个人信息页连接并确认 WorldQuant 账户身份,再创建 Key。"
/>
)}
{data && (
<section className="mcp-key-card">
<h2>连接地址</h2>
<div className="mcp-key-address">
<code>{data.endpoint}</code>
<Button onClick={() => void copy(data.endpoint)}>复制地址</Button>
</div>
<p>
客户端使用 Bearer Token 认证。Key 仅用于 MCP,不提供网页管理或正式
Alpha 提交权限。
</p>
</section>
)}
{created ? (
<section
className="mcp-key-card mcp-key-secret"
aria-label="新创建的 MCP Key"
>
<h2>Key 已创建</h2>
<p>
明文仅在这里显示一次。请保存到客户端凭据配置中;关闭、刷新或离开此页后无法再次查看。
</p>
<Input aria-label="新 Key 明文" value={created.token} readOnly />
<div className="mcp-key-actions">
<Button onClick={() => void copy(created.token)}>复制 Key</Button>
<Button onClick={() => setCreated(null)}>已保存,关闭明文</Button>
</div>
</section>
) : (
<form className="mcp-key-card" onSubmit={(event) => void create(event)}>
<h2>创建 MCP Key</h2>
<div className="mcp-key-fields">
<label>
名称
<Input
aria-label="Key 名称"
value={name}
maxLength={100}
onChange={setName}
placeholder="例如 Codex 研究助手"
/>
</label>
<label>
有效期(天)
<input
aria-label="有效期(天)"
type="number"
min="1"
max="365"
step="1"
required
value={days}
onChange={(e) => setDays(e.target.value)}
/>
</label>
</div>
<fieldset>
<legend>访问权限</legend>
<div className="mcp-key-permissions">
{scopes.map(([value, label, description]) => (
<label key={value}>
<input
type="checkbox"
checked={permissions.includes(value)}
disabled={value === "research:read" || busy}
onChange={(e) =>
setPermissions((old) =>
e.target.checked
? [...old, value]
: old.filter((item) => item !== value),
)
}
/>
<span>
<strong>{label}</strong>
<small>
{description}
{value === "research:read" ? "(必选)" : ""}
</small>
</span>
</label>
))}
</div>
</fieldset>
<Button
htmlType="submit"
theme="solid"
loading={busy}
disabled={
!data?.can_create ||
!name.trim() ||
!Number.isInteger(Number(days)) ||
Number(days) < 1 ||
Number(days) > 365
}
>
创建 Key
</Button>
</form>
)}
<section
className="mcp-key-card"
aria-label="Key 列表"
aria-busy={loading}
>
<h2>
已创建的 Key <span className="mcp-key-count">{data?.total ?? 0}</span>
</h2>
<div className="mcp-key-table">
<table>
<thead>
<tr>
<th>名称</th>
<th>权限</th>
<th>状态</th>
<th>创建时间</th>
<th>到期时间</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{data?.items.map((item) => (
<tr key={item.id}>
<td>
<strong>{item.name}</strong>
<small>{item.id}</small>
</td>
<td>
{item.scopes.map((scope) => (
<small key={scope}>
{scopes.find(([value]) => value === scope)?.[1] ??
scope}
</small>
))}
</td>
<td>{labels[item.status]}</td>
<td>{formatTime(item.created_at)}</td>
<td>{formatTime(item.expires_at)}</td>
<td>
{item.revoked_at ? (
"—"
) : revoking === item.id ? (
<div>
<p>撤销后新请求立即失效,已受理回测继续执行。</p>
<Button
type="danger"
disabled={busy}
onClick={() => void revoke(item.id)}
>
确认撤销
</Button>
<Button
disabled={busy}
onClick={() => setRevoking(null)}
>
取消
</Button>
</div>
) : (
<Button
type="danger"
disabled={busy}
onClick={() => setRevoking(item.id)}
>
撤销
</Button>
)}
</td>
</tr>
))}
</tbody>
</table>
</div>
{!loading && !data?.items.length && <p>暂无 MCP Key。</p>}
<div className="mcp-key-actions">
<Button
disabled={offset === 0 || loading || busy}
onClick={() => setOffset((v) => Math.max(0, v - 25))}
>
上一页
</Button>
<span>第 {Math.floor(offset / 25) + 1} 页</span>
<Button
disabled={!data?.has_more || loading || busy}
onClick={() => setOffset((v) => v + 25)}
>
下一页
</Button>
</div>
</section>
</section>
);
}
+118
View File
@@ -0,0 +1,118 @@
.mcp-keys-page {
overflow: auto;
width: 100%;
padding: 24px;
display: flex;
flex-direction: column;
gap: 20px;
}
.mcp-keys-page > header,
.mcp-key-address {
display: flex;
align-items: center;
justify-content: space-between;
gap: 16px;
}
.mcp-keys-page h1 {
margin: 0 0 8px;
font-size: 24px;
}
.mcp-keys-page h2 {
margin: 0 0 16px;
font-size: 16px;
}
.mcp-keys-page p {
color: var(--semi-color-text-2);
line-height: 1.6;
}
.mcp-key-card {
padding: 20px;
border: 1px solid var(--semi-color-border);
border-radius: 10px;
background: var(--semi-color-bg-1);
}
.mcp-key-fields {
display: flex;
flex-wrap: wrap;
gap: 24px;
margin-bottom: 20px;
}
.mcp-key-fields > label {
display: flex;
flex-direction: column;
gap: 8px;
min-width: 200px;
}
.mcp-key-fields input[type="number"] {
padding: 6px 12px;
border: 1px solid var(--semi-color-border);
border-radius: 4px;
background: var(--semi-color-fill-0);
color: var(--semi-color-text-0);
font: inherit;
}
.mcp-key-card fieldset {
border: 0;
padding: 0;
margin: 0 0 20px;
}
.mcp-key-card legend {
margin-bottom: 12px;
}
.mcp-key-permissions {
display: grid;
grid-template-columns: repeat(auto-fit, minmax(220px, 1fr));
gap: 16px;
}
.mcp-key-permissions label {
display: flex;
gap: 8px;
align-items: flex-start;
cursor: pointer;
}
.mcp-key-card small {
display: block;
margin-top: 4px;
color: var(--semi-color-text-2);
}
.mcp-key-actions {
display: flex;
align-items: center;
gap: 12px;
margin-top: 16px;
}
.mcp-key-table {
overflow: auto;
}
.mcp-key-table table {
width: 100%;
border-collapse: collapse;
text-align: left;
}
.mcp-key-table th,
.mcp-key-table td {
padding: 12px;
border-bottom: 1px solid var(--semi-color-border);
vertical-align: top;
}
.mcp-key-table th {
color: var(--semi-color-text-2);
font-weight: 500;
white-space: nowrap;
}
.mcp-key-count {
color: var(--semi-color-text-2);
font-weight: 400;
}
.mcp-key-address code {
overflow-wrap: anywhere;
}
@media (max-width: 700px) {
.mcp-keys-page {
padding: 12px;
}
.mcp-key-address {
align-items: flex-start;
flex-direction: column;
}
}
+1
View File
@@ -7,6 +7,7 @@ export const sourceLabel = (kind: string) =>
({
chatbox: "Chatbox 研究",
manual: "手工研究",
mcp: "MCP 研究",
ai: "AI 研究(历史)",
template: "模板研究",
variant: "Alpha 变体",
+86 -1
View File
@@ -28,7 +28,9 @@ test("draft, immutable preview, mixed-result persistence and responsive workspac
await page.getByRole("button", { name: "新建回测" }).click();
await page.getByLabel("运行名称", { exact: true }).fill("浏览器回测验收");
await page.getByRole("textbox", { name: "Region", exact: true }).fill("USA");
await page.getByRole("textbox", { name: "Universe", exact: true }).fill("TOP3000");
await page
.getByRole("textbox", { name: "Universe", exact: true })
.fill("TOP3000");
await page.getByLabel("回测候选").fill("rank(close)\n-rank(volume)");
await page.getByRole("button", { name: "保存草稿", exact: true }).click();
await expect(page.getByText("草稿已保存", { exact: true })).toBeVisible();
@@ -103,3 +105,86 @@ test("AI prepares one fixed preview, confirms once, and shows live run independe
await chat.getByRole("button", { name: "打开回测详情", exact: true }).click();
await expect(page.getByText("1 / 1 已保存", { exact: true })).toBeVisible();
});
test("MCP direct batch appears through its run link and source filter", async ({
page,
}) => {
await login(page);
const token = (
await (
await page.request.post("/api/v1/test/mcp-token", { headers })
).json()
).token;
const response = await page.request.post("/api/v1/mcp/", {
headers: {
Authorization: `Bearer ${token}`,
Accept: "application/json, text/event-stream",
},
data: {
jsonrpc: "2.0",
id: "browser-batch",
method: "tools/call",
params: {
name: "submit_backtests",
arguments: {
name: "MCP 浏览器留痕",
idempotency_key: "browser-mcp",
duplicate_policy: "rerun",
candidates: [
{
client_item_id: "browser-1",
expression: "rank(close) + 987",
settings: {
instrumentType: "EQUITY",
region: "USA",
universe: "TOP3000",
delay: 1,
decay: 0,
neutralization: "INDUSTRY",
truncation: 0.08,
pasteurization: "ON",
unitHandling: "VERIFY",
nanHandling: "OFF",
language: "FASTEXPR",
visualization: false,
maxTrade: "OFF",
maxPosition: "OFF",
},
},
],
source: {
reference: "browser-conversation",
hypothesis: "验证直接回测留痕",
},
},
},
},
});
expect(response.ok()).toBeTruthy();
const result = (await response.json()).result;
expect(result.isError).toBeFalsy();
const run = result.structuredContent;
await page.goto(run.web_url);
await expect(page.getByText("1 / 1 已保存", { exact: true })).toBeVisible({
timeout: 20000,
});
await expect(
page.getByText("研究来源:MCP 研究", { exact: true }),
).toBeVisible();
await expect(
page.getByText("来源引用:browser-conversation", { exact: true }),
).toBeVisible();
await page.reload();
await expect(page.getByText("1 / 1 已保存", { exact: true })).toBeVisible();
await page.keyboard.press("Escape");
await page
.getByRole("dialog")
.getByRole("heading")
.getByRole("button")
.click();
await page.getByText("全部研究来源", { exact: true }).click();
await page.getByRole("option", { name: /MCP 研究/ }).click();
await expect(
page.getByRole("button", { name: "MCP 浏览器留痕", exact: true }),
).toBeVisible();
});
+60
View File
@@ -0,0 +1,60 @@
import { expect, test } from "@playwright/test";
test("MCP Key menu creates, hides secrets on navigation, and revokes", async ({
page,
}) => {
await page.goto("/#mcp-keys");
await page.getByLabel("密码", { exact: true }).fill("browser-test-password");
await page.getByRole("button", { name: "进入工作空间" }).click();
await expect(
page.getByRole("heading", { name: "MCP Key", exact: true }),
).toBeVisible();
const headers = { "X-WQ-Request": "1" };
await page.request.put("/api/v1/account/credentials", {
headers,
data: { email: "test@example.com", password: "synthetic-password" },
});
await page.request.post("/api/v1/account/connect", { headers });
await expect
.poll(
async () =>
(await (await page.request.get("/api/v1/account")).json())
.connection_status,
)
.toBe("connected");
await page.getByRole("button", { name: "刷新列表", exact: true }).click();
await page.getByLabel("Key 名称").fill("Browser test reader");
await page.getByRole("button", { name: "创建 Key", exact: true }).click();
await expect(page.getByLabel("新 Key 明文")).toBeVisible();
const secret = await page.getByLabel("新 Key 明文").inputValue();
expect(secret).toMatch(/^wqmcp_/);
const listed = await (await page.request.get("/api/v1/mcp-tokens")).json();
expect(JSON.stringify(listed)).not.toContain(secret);
const item = listed.items.find(
(row: { name: string }) => row.name === "Browser test reader",
);
expect(item.scopes).toEqual(["research:read"]);
await page.screenshot({
path: "test-results/mcp-key-page.png",
fullPage: true,
mask: [page.getByLabel("新 Key 明文")],
});
await page
.getByRole("navigation", { name: "主导航", exact: true })
.getByRole("button", { name: "数据目录", exact: true })
.click();
await page
.getByRole("navigation", { name: "主导航", exact: true })
.getByRole("button", { name: "MCP Key", exact: true })
.click();
await expect(page.getByLabel("新 Key 明文")).toHaveCount(0);
const row = page.getByRole("row").filter({ hasText: "Browser test reader" });
await row.getByRole("button", { name: "撤销", exact: true }).click();
await row.getByRole("button", { name: "确认撤销", exact: true }).click();
await expect(row).toContainText("已撤销");
await page.reload();
await expect(
page.getByRole("heading", { name: "MCP Key", exact: true }),
).toBeVisible();
await expect(page.getByLabel("新 Key 明文")).toHaveCount(0);
});
+7 -1
View File
@@ -4,7 +4,13 @@ import react from "@vitejs/plugin-react";
export default defineConfig({
plugins: [react()],
server: {
proxy: { "/api": process.env.WQ_DEV_API ?? "http://127.0.0.1:8000" },
proxy: {
"/api": {
target: process.env.WQ_DEV_API ?? "http://127.0.0.1:8000",
// Match production proxy behavior so MCP validates the public Host.
changeOrigin: false,
},
},
},
build: { chunkSizeWarningLimit: 1500 },
});