feat(mcp-server): Heicode magic-link 邮箱登录三端点（仅登录，不触碰自有登录）

按 Docs/Heicode-magic-link…md 契约 §7.4/§9.4/§11/§13.2 新增并行登录方式：
- app/magic_link.py：Redis 一次性 token(600s)/code(120s) + 邮箱60s限流 + 复用
  现有 SMTP 通道发链接邮件（SMTP_PASSWORD 未配则不外发，DEBUG 打日志）
- app/routes/auth.py：新增 /api/auth/magic-link/{request,landing,verify}
  · request：IP+邮箱限流，防枚举一视同仁，仅对已存在 role=user 发信(D-1/D-2/D-3)
  · landing：消费 token→生成 code→302 heicode://auth/callback，失败回 HTML
  · verify：消费 code→复用 create_access_token/refresh + 与 /login 逐字段相同
    token_data → 登录产物等价，EU/计费零改动(§1.3)
- config.py：新增 MAGIC_LINK_PUBLIC_BASE_URL（默认 APIM 域，§11 终态）
- 三端点不声明 Depends(require_auth) 即公开，未改 allow_paths(§12.2)
- /login、/me、/refresh、/logout、/register 一行未改

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

services/mcp-server/app/magic_link.py

@@ -0,0 +1,250 @@
+"""
+Heicode magic-link 邮箱登录 —— Redis 一次性 token/code 存取 + 登录链接邮件发送
+
+本模块为 Heicode magic-link 登录新增的独立支撑层，契约见
+`Docs/Heicode-magic-link邮箱登录-给mcp-server的对接需求.md`（§2/§7/§9/§11/§12）。
+
+设计要点（与既有逻辑零耦合，绝不改动密码登录/注册验证码链路）：
+- **复用** `email_verification` 的 SMTP 通道（smtp.189.cn / taijiagent@189.cn）发送
+  登录链接邮件；
+- **复用** `state.redis_client` 存一次性凭证，沿用现有验证码同款 one-time 模式
+  （`setex` 落地 + 命中即 `delete`，含 Redis 集群 MOVED 重定向重试）；
+- token / code 均为短 TTL、一次性：
+    - magic-link token：TTL 600s（邮件链接里的凭证）
+    - 一次性 code：TTL 120s（landing 校验通过后换取登录态的凭证）
+"""
+
+from __future__ import annotations
+
+import json
+import secrets
+from typing import Optional
+
+import structlog
+
+from app.state import get_state
+# 复用现有 SMTP 通道与同步发信实现，不另起炉灶
+from app.email_verification import (
+    SMTP_EMAIL,
+    SMTP_PASSWORD,
+    _send_email_sync,
+)
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+
+logger = structlog.get_logger(__name__)
+
+# ===== TTL / Redis key 约定（契约 §2 / §7.1）=====
+MAGIC_LINK_TOKEN_TTL_SECONDS = 600   # 邮件链接 token：10 分钟
+MAGIC_LINK_CODE_TTL_SECONDS = 120    # 一次性 code：≤2 分钟
+
+_TOKEN_KEY_PREFIX = "magic_link_token:"
+_CODE_KEY_PREFIX = "magic_link_code:"
+# 邮箱维度限流（§13.2「邮箱限流」）。**独立**于注册/忘记密码共用的
+# verification_rate_limit:{email}，避免 magic-link 与那两条流程互相误伤。
+_EMAIL_RATE_LIMIT_KEY_PREFIX = "magic_link_rate_limit:"
+EMAIL_RATE_LIMIT_SECONDS = 60  # 同一邮箱 60s 内只接受一次申请
+
+_MAX_REDIS_RETRIES = 3
+
+
+def generate_magic_link_token() -> str:
+    """生成 magic-link token（URL-safe，随邮件链接下发）。"""
+    return secrets.token_urlsafe(32)
+
+
+def generate_one_time_code() -> str:
+    """生成一次性 code（URL-safe，landing 302 回跳给客户端）。"""
+    return secrets.token_urlsafe(24)
+
+
+async def _redis_get(key: str) -> Optional[str]:
+    """带 Redis 集群 MOVED 重定向重试的 GET（与 email_verification 同款）。"""
+    state = get_state()
+    if not state.redis_client:
+        logger.warning("magic_link_redis_unavailable", op="get")
+        return None
+    for attempt in range(_MAX_REDIS_RETRIES):
+        try:
+            return await state.redis_client.get(key)
+        except Exception as e:  # noqa: BLE001
+            if "MOVED" in str(e) and attempt < _MAX_REDIS_RETRIES - 1:
+                import asyncio
+                await asyncio.sleep(0.1)
+                continue
+            raise
+    return None
+
+
+async def _redis_delete(key: str) -> None:
+    """带 MOVED 重试的 DELETE；删除失败不致命（一次性消费已凭 get 判定）。"""
+    state = get_state()
+    if not state.redis_client:
+        return
+    for attempt in range(_MAX_REDIS_RETRIES):
+        try:
+            await state.redis_client.delete(key)
+            return
+        except Exception as e:  # noqa: BLE001
+            if "MOVED" in str(e) and attempt < _MAX_REDIS_RETRIES - 1:
+                import asyncio
+                await asyncio.sleep(0.1)
+                continue
+            logger.warning("magic_link_redis_delete_failed", key_prefix=key[:24], error=str(e))
+            return
+
+
+async def _setex(key: str, ttl: int, value: str) -> bool:
+    state = get_state()
+    if not state.redis_client:
+        logger.warning("magic_link_redis_unavailable", op="setex")
+        return False
+    try:
+        await state.redis_client.setex(key, ttl, value)
+        return True
+    except Exception as e:  # noqa: BLE001
+        logger.error("magic_link_redis_setex_failed", key_prefix=key[:24], error=str(e))
+        return False
+
+
+# ===== 邮箱维度限流（§13.2）=====
+
+async def check_email_rate_limit(email: str) -> tuple[bool, int]:
+    """检查同一邮箱的申请频率限制。
+
+    Returns: (是否可以申请, 剩余等待秒数)。Redis 不可用时放行（不阻塞用户），
+    与 email_verification.check_rate_limit 的容错口径一致。
+    """
+    state = get_state()
+    if not state.redis_client:
+        return True, 0
+    try:
+        ttl = await state.redis_client.ttl(_EMAIL_RATE_LIMIT_KEY_PREFIX + email)
+        if ttl and ttl > 0:
+            return False, ttl
+        return True, 0
+    except Exception as e:  # noqa: BLE001
+        logger.error("magic_link_email_rate_limit_check_failed", email=email, error=str(e))
+        return True, 0
+
+
+async def set_email_rate_limit(email: str) -> None:
+    """对该邮箱设置 60s 申请冷却。**对存在/不存在的邮箱一律设置**——否则
+    「存在→后续 429 / 不存在→后续 200」会泄漏邮箱存在性，破坏防枚举（D-2）。"""
+    state = get_state()
+    if not state.redis_client:
+        return
+    try:
+        await state.redis_client.setex(
+            _EMAIL_RATE_LIMIT_KEY_PREFIX + email, EMAIL_RATE_LIMIT_SECONDS, "1"
+        )
+    except Exception as e:  # noqa: BLE001
+        logger.error("magic_link_email_rate_limit_set_failed", email=email, error=str(e))
+
+
+# ===== token：email + state 绑定 =====
+
+async def store_magic_link_token(token: str, email: str, state: str) -> bool:
+    """存 magic-link token → {email, state}，TTL 600s。"""
+    payload = json.dumps({"email": email, "state": state})
+    return await _setex(_TOKEN_KEY_PREFIX + token, MAGIC_LINK_TOKEN_TTL_SECONDS, payload)
+
+
+async def consume_magic_link_token(token: str) -> Optional[dict]:
+    """一次性消费 token：命中则返回 {email, state} 并删除；否则 None。"""
+    if not token:
+        return None
+    key = _TOKEN_KEY_PREFIX + token
+    raw = await _redis_get(key)
+    if not raw:
+        return None
+    await _redis_delete(key)
+    try:
+        return json.loads(raw)
+    except (ValueError, TypeError):
+        logger.warning("magic_link_token_payload_corrupt")
+        return None
+
+
+# ===== code：user_id + email + state 绑定 =====
+
+async def store_one_time_code(code: str, user_id: str, email: str, state: str) -> bool:
+    """存一次性 code → {user_id, email, state}，TTL ≤120s。"""
+    payload = json.dumps({"user_id": user_id, "email": email, "state": state})
+    return await _setex(_CODE_KEY_PREFIX + code, MAGIC_LINK_CODE_TTL_SECONDS, payload)
+
+
+async def consume_one_time_code(code: str) -> Optional[dict]:
+    """一次性消费 code：命中则返回 {user_id, email, state} 并删除；否则 None。"""
+    if not code:
+        return None
+    key = _CODE_KEY_PREFIX + code
+    raw = await _redis_get(key)
+    if not raw:
+        return None
+    await _redis_delete(key)
+    try:
+        return json.loads(raw)
+    except (ValueError, TypeError):
+        logger.warning("magic_link_code_payload_corrupt")
+        return None
+
+
+# ===== 登录链接邮件（复用现有 SMTP 通道）=====
+
+async def send_magic_link_email(email: str, link_url: str) -> bool:
+    """发送 magic-link 登录链接邮件。
+
+    复用 email_verification 的 SMTP 通道。SMTP_PASSWORD 未注入时（邮件基建尚未
+    就位，契约 §12.4「默认不外发」），记录链接到日志并返回 False，不抛异常 —— 让
+    上层 `request` 端点照常返回 200（防枚举）。
+    """
+    if not SMTP_PASSWORD:
+        # 邮件基建未就位：不外发，仅在 DEBUG 下打日志便于联调对 mock。
+        import os
+        if os.getenv("DEBUG", "false").lower() == "true" or \
+                os.getenv("ENABLE_TEST_MODE", "false").lower() == "true":
+            logger.warning(
+                "magic_link_email_not_sent_smtp_unconfigured",
+                email=email,
+                link_url=link_url,
+                hint="SMTP_PASSWORD 未配置；测试模式下仅打印链接，不真正发信",
+            )
+        else:
+            logger.error("magic_link_email_smtp_unconfigured", email=email)
+        return False
+
+    try:
+        msg = MIMEMultipart()
+        msg["From"] = SMTP_EMAIL
+        msg["To"] = email
+        msg["Subject"] = "Taiji AI-PAD 登录链接"
+        body = f"""
+尊敬的用户：
+
+您正在登录 HeiCode 客户端。请在 10 分钟内，**在已安装 HeiCode 的同一台设备上**
+点击下面的链接完成登录：
+
+{link_url}
+
+提示：此链接仅用于本次登录，点击后会自动回跳到本机 HeiCode 客户端。
+请务必在安装了 HeiCode 的同一台设备上打开此链接，否则客户端无法收到回跳。
+
+如果您没有发起登录，请忽略此邮件，您的账户仍然安全。
+
+此邮件由系统自动发送，请勿回复。
+
+---
+Taiji AI-PAD 团队
+"""
+        msg.attach(MIMEText(body, "plain", "utf-8"))
+
+        import asyncio
+        loop = asyncio.get_event_loop()
+        await loop.run_in_executor(None, _send_email_sync, msg)
+        logger.info("magic_link_email_sent", email=email)
+        return True
+    except Exception as e:  # noqa: BLE001
+        logger.error("magic_link_email_send_failed", email=email, error=str(e),
+                     error_type=type(e).__name__)
+        return False

services/mcp-server/app/routes/auth.py

@@ -4,6 +4,7 @@
 
 from datetime import timedelta
 from fastapi import APIRouter, Depends, HTTPException, status, Query, Request
+from fastapi.responses import RedirectResponse, HTMLResponse
 from sqlalchemy import select, and_
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.exc import IntegrityError
@@ -37,7 +38,19 @@ from app.schemas import (
 )
 from app.email_verification import verify_code, peek_verification_code, send_and_store_verification_code, check_rate_limit, send_password_reset_code
 from app.audit import log_audit_event
-from pydantic import BaseModel
+from app.magic_link import (
+    generate_magic_link_token,
+    generate_one_time_code,
+    store_magic_link_token,
+    consume_magic_link_token,
+    store_one_time_code,
+    consume_one_time_code,
+    send_magic_link_email,
+    check_email_rate_limit,
+    set_email_rate_limit,
+    MAGIC_LINK_TOKEN_TTL_SECONDS,
+)
+from pydantic import BaseModel, EmailStr
 from app.agent_manager_client import get_agent_manager_client, AgentManagerError
 from app.litellm_client import get_litellm_client, LiteLLMClientError
 from config import settings
@@ -1317,3 +1330,215 @@ async def set_billing_provider(
         "old_billing_provider": old,
     })
 
+
+# ============================================================
+# Heicode magic-link 邮箱登录（§2 契约 / §8 D-1·D-2·D-3 / §11 D-4=APIM / §12）
+# —— 纯新增的并行登录方式：不输密码，邮箱收链接，点链接回跳客户端完成登录。
+# 绝不改动 /login、/me、/refresh、/logout、/register 任何现有行为（满足 §1.1）。
+#   · D-1 仅登录：verify 只对**已存在 user** 签发登录产物，绝不触发 register provisioning；
+#                未注册邮箱在 request 阶段静默不发信。
+#   · D-2 防枚举：request 无论邮箱是否注册一律返回成功。
+#   · D-3 仅 role=user：channel/admin 继续走密码登录。
+#   · §1.3 登录产物等价：verify 复用 create_access_token/create_refresh_token
+#                + 与 /login 逐字段相同的 token_data → EU/计费零改动。
+# 三端点均**不声明 Depends(require_auth)** 即公开（§12.2 已证实，无需改 allow_paths）。
+# ============================================================
+
+# 与密码登录独立的限流计数器（IP 维度 5 次/60s，对齐 §2.1）
+_magic_link_rate_limit = _LoginRateLimit()
+
+
+class _MagicLinkRequest(BaseModel):
+    """magic-link/request 请求体。"""
+    email: EmailStr
+
+
+class _MagicLinkVerify(BaseModel):
+    """magic-link/verify 请求体。device_pubkey 可选，我方忽略（设备配对在 HM 侧，
+    见契约 §3 / §7.1 D-4），且**不改变** token 结构。"""
+    code: str
+    state: str
+    device_pubkey: Optional[str] = None
+
+
+_LANDING_INVALID_HTML = """<!DOCTYPE html>
+<html lang="zh-CN"><head><meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>链接无效</title></head>
+<body style="font-family:system-ui,-apple-system,Segoe UI,sans-serif;max-width:520px;margin:80px auto;padding:0 24px;color:#222;text-align:center">
+<h2>登录链接无效或已过期</h2>
+<p>请回到 HeiCode 客户端重新获取登录链接。</p>
+<p style="color:#888;font-size:14px">提示：请在已安装 HeiCode 的同一台设备上打开邮件链接。</p>
+</body></html>"""
+
+
+@router.post("/magic-link/request", response_model=SuccessResponse)
+async def magic_link_request(
+    req: _MagicLinkRequest,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    _: None = Depends(_magic_link_rate_limit),
+):
+    """申请 magic-link 登录链接（契约 §2.1）。
+
+    防枚举（D-2）：无论邮箱是否注册一律返回成功；仅当邮箱对应**已存在的
+    role=user 用户**（D-1 仅登录 + D-3）时才真正生成 token 并发信，其余情况静默。
+    """
+    email = req.email
+    # state 始终下发（客户端存下，回跳时严格比对）；request_id 供追踪。
+    state = secrets.token_urlsafe(16)
+    request_id = str(uuid.uuid4())
+
+    # 邮箱维度限流（§13.2）：在查用户之前判定，且对存在/不存在邮箱一视同仁，
+    # 既防止对真实用户的邮件轰炸，又不泄漏邮箱存在性（D-2 防枚举）。
+    can_send, remaining = await check_email_rate_limit(email)
+    if not can_send:
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail=f"请等待{remaining}秒后再重新申请登录链接",
+            headers={"Retry-After": str(remaining)},
+        )
+    # 无论后续是否真正发信，都先打上冷却（防枚举）
+    await set_email_rate_limit(email)
+
+    result = await db.execute(select(User).where(User.email == email))
+    user = result.scalar_one_or_none()
+
+    if user is not None and user.role == "user":
+        token = generate_magic_link_token()
+        stored = await store_magic_link_token(token, email, state)
+        if stored:
+            base = settings.magic_link_public_base_url.rstrip("/")
+            link = f"{base}/api/auth/magic-link/landing?token={token}&state={state}"
+            # 发信失败不影响响应（防枚举 + 邮件基建未就位时静默，见 §12.4）
+            await send_magic_link_email(email, link)
+        else:
+            logger.error("magic_link_token_store_failed", email=email)
+    else:
+        # 未注册 / 非 user 角色：静默不发信（与防枚举一致）
+        logger.info("magic_link_request_silent_skip", email=email)
+
+    return SuccessResponse(data={
+        "request_id": request_id,
+        "state": state,
+        "expires_in_sec": MAGIC_LINK_TOKEN_TTL_SECONDS,
+    })
+
+
+@router.get("/magic-link/landing")
+async def magic_link_landing(
+    token: str = Query(..., description="magic-link token"),
+    state: str = Query(..., description="客户端 state"),
+    db: AsyncSession = Depends(get_db),
+):
+    """邮件链接指向的落地页（契约 §2.2）。浏览器直接打开，经 APIM 反代透传（§11）。
+
+    校验 token（存在/未过期/未用，一次性消费）+ state 一致 + user 仍存在且
+    role=user → 生成一次性 code（≤2min）并 **302 跳转** 到桌面深链
+    `heicode://auth/callback?code=...&state=...`；任一校验失败返回人类可读 HTML。
+    """
+    payload = await consume_magic_link_token(token)
+    if not payload or payload.get("state") != state:
+        return HTMLResponse(content=_LANDING_INVALID_HTML, status_code=status.HTTP_400_BAD_REQUEST)
+
+    email = payload.get("email")
+    result = await db.execute(select(User).where(User.email == email))
+    user = result.scalar_one_or_none()
+    # D-3：仅 role=user；用户被删/改角色则视为无效
+    if user is None or user.role != "user":
+        return HTMLResponse(content=_LANDING_INVALID_HTML, status_code=status.HTTP_400_BAD_REQUEST)
+
+    code = generate_one_time_code()
+    stored = await store_one_time_code(code, str(user.id), email, state)
+    if not stored:
+        logger.error("magic_link_code_store_failed", email=email)
+        return HTMLResponse(content=_LANDING_INVALID_HTML, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    redirect_url = f"heicode://auth/callback?code={code}&state={state}"
+    return RedirectResponse(url=redirect_url, status_code=status.HTTP_302_FOUND)
+
+
+@router.post("/magic-link/verify", response_model=SuccessResponse)
+async def magic_link_verify(
+    req: _MagicLinkVerify,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+    _: None = Depends(_magic_link_rate_limit),
+):
+    """用一次性 code 换登录态（契约 §2.3）。
+
+    成功响应与 `POST /api/auth/login`（user 分支）**逐字段一致**：同 token_data、
+    同 create_access_token/create_refresh_token、同 24h/7d TTL、同 {token,
+    refreshToken, user{id,name,email,role,channelId}}。→ EU/计费零改动（§1.3）。
+    """
+    success = False
+    result_user_id: Optional[str] = None
+    error_msg: Optional[str] = None
+    try:
+        payload = await consume_one_time_code(req.code)
+        if not payload or payload.get("state") != req.state:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="登录凭证无效或已过期",
+            )
+
+        # 按 code 绑定的 user_id 解析（D-3：仅 role=user）
+        uid_raw = payload.get("user_id")
+        try:
+            user = await db.get(User, uuid.UUID(str(uid_raw)))
+        except (ValueError, TypeError):
+            user = None
+        if user is None or user.role != "user":
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="登录凭证无效或已过期",
+            )
+
+        # 与 /login 一致：更新最后登录时间
+        user.last_login_at = datetime.utcnow()
+        await db.commit()
+
+        # 与 /login user 分支逐字段相同的 token_data
+        token_data = {
+            "sub": str(user.id),
+            "email": user.email,
+            "role": user.role,
+            "channelId": str(user.channel_id) if user.channel_id else None,
+        }
+        access_token = create_access_token(data=token_data)
+        refresh_token = create_refresh_token(data=token_data)
+
+        success = True
+        result_user_id = str(user.id)
+        return SuccessResponse(
+            data={
+                "token": access_token,
+                "refreshToken": refresh_token,
+                "user": {
+                    "id": str(user.id),
+                    "name": user.name or user.full_name,
+                    "email": user.email,
+                    "role": user.role,
+                    "channelId": str(user.channel_id) if user.channel_id else None,
+                },
+            }
+        )
+    except HTTPException as e:
+        error_msg = e.detail if isinstance(e.detail, str) else str(e.detail)
+        raise
+    finally:
+        try:
+            await log_audit_event(
+                action="auth.login",
+                resource_type="user",
+                resource_id=result_user_id,
+                user_id=result_user_id,
+                success=success,
+                details={"role": "user", "method": "magic_link"},
+                error_message=error_msg,
+                request=request,
+                db=db,
+            )
+        except Exception as audit_exc:
+            logger.warning("magic_link_verify_audit_failed", error=str(audit_exc))
+

services/mcp-server/config.py

@@ -136,6 +136,15 @@ class Settings(BaseSettings):
     # 用 Authorization: Bearer <这个值> 鉴权
     heicode_internal_service_token: str = os.getenv("HEICODE_INTERNAL_SERVICE_TOKEN", "")
 
+    # Heicode magic-link 邮箱登录：本服务对外公网基址（见
+    # Docs/Heicode-magic-link邮箱登录-给mcp-server的对接需求.md §11 拍板=统一走 APIM）。
+    # 用于在邮件正文里拼出用户浏览器可直接打开的 landing 绝对 URL：
+    #   {base}/api/auth/magic-link/landing?token=...&state=...
+    # 由运维注入；缺省取 §11 定稿的 APIM 域。
+    magic_link_public_base_url: str = os.getenv(
+        "MAGIC_LINK_PUBLIC_BASE_URL", "https://apimtaiji.azure-api.net/api/mcp"
+    )
+
     # 云存储设置（Azure Blob Storage）
     azure_storage_connection_string: str = os.getenv("AZURE_STORAGE_CONNECTION_STRING", "")
     s3_bucket: str = os.getenv("S3_BUCKET", "taiji-ai-exports")