feat: support new model Grok 4.20 Beta

app/services/grok/batch_services/usage.py

@@ -46,7 +46,11 @@ async def get(self, token: str) -> Dict:
                 async with AsyncSession() as session:
                     response = await RateLimitsReverse.request(session, token)
                 data = response.json()
-                remaining = data.get("remainingTokens", 0)
+                remaining = data.get("remainingTokens")
+                if remaining is None:
+                    remaining = data.get("remainingQueries")
+                    if remaining is not None:
+                        data["remainingTokens"] = remaining
                 logger.info(
                     f"Usage sync success: remaining={remaining}, token={token[:10]}..."
                 )

app/services/grok/services/chat.py

@@ -33,6 +33,63 @@
 _CHAT_SEM_VALUE = None
 
 
+def extract_tool_text(raw: str, rollout_id: str = "") -> str:
+    if not raw:
+        return ""
+    name_match = re.search(
+        r"<xai:tool_name>(.*?)</xai:tool_name>", raw, flags=re.DOTALL
+    )
+    args_match = re.search(
+        r"<xai:tool_args>(.*?)</xai:tool_args>", raw, flags=re.DOTALL
+    )
+
+    name = name_match.group(1) if name_match else ""
+    if name:
+        name = re.sub(r"<!\[CDATA\[(.*?)\]\]>", r"\1", name, flags=re.DOTALL).strip()
+
+    args = args_match.group(1) if args_match else ""
+    if args:
+        args = re.sub(r"<!\[CDATA\[(.*?)\]\]>", r"\1", args, flags=re.DOTALL).strip()
+
+    payload = None
+    if args:
+        try:
+            payload = orjson.loads(args)
+        except orjson.JSONDecodeError:
+            payload = None
+
+    label = name
+    text = args
+    prefix = f"[{rollout_id}]" if rollout_id else ""
+
+    if name == "web_search":
+        label = f"{prefix}[WebSearch]"
+        if isinstance(payload, dict):
+            text = payload.get("query") or payload.get("q") or ""
+    elif name == "search_images":
+        label = f"{prefix}[SearchImage]"
+        if isinstance(payload, dict):
+            text = (
+                payload.get("image_description")
+                or payload.get("description")
+                or payload.get("query")
+                or ""
+            )
+    elif name == "chatroom_send":
+        label = f"{prefix}[AgentThink]"
+        if isinstance(payload, dict):
+            text = payload.get("message") or ""
+
+    if label and text:
+        return f"{label} {text}".strip()
+    if label:
+        return label
+    if text:
+        return text
+    # Fallback: strip tags to keep any raw text.
+    return re.sub(r"<[^>]+>", "", raw, flags=re.DOTALL).strip()
+
+
 def _get_chat_semaphore() -> asyncio.Semaphore:
     global _CHAT_SEMAPHORE, _CHAT_SEM_VALUE
     value = max(1, int(get_config("chat.concurrent")))
@@ -115,7 +172,7 @@ async def chat(
         self,
         token: str,
         message: str,
-        model: str = "grok-3",
+        model: str,
         mode: str = None,
         stream: bool = None,
         file_attachments: List[str] = None,
@@ -338,18 +395,86 @@ def __init__(self, model: str, token: str = "", show_think: bool = None):
         super().__init__(model, token)
         self.response_id: str = None
         self.fingerprint: str = ""
+        self.rollout_id: str = ""
         self.think_opened: bool = False
         self.role_sent: bool = False
         self.filter_tags = get_config("app.filter_tags")
+        self.tool_usage_enabled = (
+            "xai:tool_usage_card" in (self.filter_tags or [])
+        )
+        self._tool_usage_opened = False
+        self._tool_usage_buffer = ""
 
         self.show_think = bool(show_think)
 
+    def _filter_tool_card(self, token: str) -> str:
+        if not token or not self.tool_usage_enabled:
+            return token
+
+        output_parts: list[str] = []
+        rest = token
+        start_tag = "<xai:tool_usage_card"
+        end_tag = "</xai:tool_usage_card>"
+
+        while rest:
+            if self._tool_usage_opened:
+                end_idx = rest.find(end_tag)
+                if end_idx == -1:
+                    self._tool_usage_buffer += rest
+                    return "".join(output_parts)
+                end_pos = end_idx + len(end_tag)
+                self._tool_usage_buffer += rest[:end_pos]
+                line = extract_tool_text(self._tool_usage_buffer, self.rollout_id)
+                if line:
+                    if output_parts and not output_parts[-1].endswith("\n"):
+                        output_parts[-1] += "\n"
+                    output_parts.append(f"{line}\n")
+                self._tool_usage_buffer = ""
+                self._tool_usage_opened = False
+                rest = rest[end_pos:]
+                continue
+
+            start_idx = rest.find(start_tag)
+            if start_idx == -1:
+                output_parts.append(rest)
+                break
+
+            if start_idx > 0:
+                output_parts.append(rest[:start_idx])
+
+            end_idx = rest.find(end_tag, start_idx)
+            if end_idx == -1:
+                self._tool_usage_opened = True
+                self._tool_usage_buffer = rest[start_idx:]
+                break
+
+            end_pos = end_idx + len(end_tag)
+            raw_card = rest[start_idx:end_pos]
+            line = extract_tool_text(raw_card, self.rollout_id)
+            if line:
+                if output_parts and not output_parts[-1].endswith("\n"):
+                    output_parts[-1] += "\n"
+                output_parts.append(f"{line}\n")
+            rest = rest[end_pos:]
+
+        return "".join(output_parts)
+
     def _filter_token(self, token: str) -> str:
         """Filter special tags in current token only."""
-        if not self.filter_tags or not token:
+        if not token:
+            return token
+
+        if self.tool_usage_enabled:
+            token = self._filter_tool_card(token)
+            if not token:
+                return ""
+
+        if not self.filter_tags:
             return token
 
         for tag in self.filter_tags:
+            if tag == "xai:tool_usage_card":
+                continue
             if f"<{tag}" in token or f"</{tag}" in token:
                 return ""
 
@@ -408,6 +533,8 @@ async def process(self, response: AsyncIterable[bytes]) -> AsyncGenerator[str, N
                     self.fingerprint = llm.get("modelHash", "")
                 if rid := resp.get("responseId"):
                     self.response_id = rid
+                if rid := resp.get("rolloutId"):
+                    self.rollout_id = str(rid)
 
                 if not self.role_sent:
                     yield self._sse(role="assistant")
@@ -537,7 +664,28 @@ def _filter_content(self, content: str) -> str:
             return content
 
         result = content
+        if "xai:tool_usage_card" in self.filter_tags:
+            rollout_id = ""
+            rollout_match = re.search(
+                r"<rolloutId>(.*?)</rolloutId>", result, flags=re.DOTALL
+            )
+            if rollout_match:
+                rollout_id = rollout_match.group(1).strip()
+
+            result = re.sub(
+                r"<xai:tool_usage_card[^>]*>.*?</xai:tool_usage_card>",
+                lambda match: (
+                    f"{extract_tool_text(match.group(0), rollout_id)}\n"
+                    if extract_tool_text(match.group(0), rollout_id)
+                    else ""
+                ),
+                result,
+                flags=re.DOTALL,
+            )
+
         for tag in self.filter_tags:
+            if tag == "xai:tool_usage_card":
+                continue
             pattern = rf"<{re.escape(tag)}[^>]*>.*?</{re.escape(tag)}>|<{re.escape(tag)}[^>]*/>"
             result = re.sub(pattern, "", result, flags=re.DOTALL)
 

app/services/grok/services/model.py

@@ -163,6 +163,17 @@ class ModelService:
             is_image_edit=False,
             is_video=False,
         ),
+        ModelInfo(
+            model_id="grok-4.20-beta",
+            grok_model="grok-420",
+            model_mode="MODEL_MODE_GROK_420",
+            tier=Tier.BASIC,
+            cost=Cost.LOW,
+            display_name="GROK-4.20-BETA",
+            is_image=False,
+            is_image_edit=False,
+            is_video=False,
+        ),
         ModelInfo(
             model_id="grok-imagine-1.0",
             grok_model="grok-3",

app/services/token/manager.py

@@ -375,8 +375,12 @@ async def sync_usage(
             result = await usage_service.get(token_str)
 
             if result and "remainingTokens" in result:
+                new_quota = result.get("remainingTokens")
+                if new_quota is None:
+                    new_quota = result.get("remainingQueries")
+                if new_quota is None:
+                    return False
                 old_quota = target_token.quota
-                new_quota = result["remainingTokens"]
 
                 target_token.update_quota(new_quota)
                 target_token.record_success(is_usage=is_usage)
@@ -699,7 +703,11 @@ async def _refresh_one(token_info: TokenInfo) -> dict:
                         result = await usage_service.get(token_str)
 
                         if result and "remainingTokens" in result:
-                            new_quota = result["remainingTokens"]
+                            new_quota = result.get("remainingTokens")
+                            if new_quota is None:
+                                new_quota = result.get("remainingQueries")
+                            if new_quota is None:
+                                return {"recovered": False, "expired": False}
                             old_quota = token_info.quota
                             old_status = token_info.status
 

app/static/common/js/toast.js

@@ -45,3 +45,34 @@ function showToast(message, type = 'success') {
     });
   }, 3000);
 }
+
+(function showRateLimitNoticeOnce() {
+  const noticeKey = 'grok2api_rate_limits_notice_v1';
+  const noticeText = 'GROK官方服务 rate-limits 更新后暂时无法准确计算 Token 剩余，等待官方接口优化后持续修复';
+  const path = window.location.pathname || '';
+
+  if (!path.startsWith('/admin') || path.startsWith('/admin/login')) {
+    return;
+  }
+
+  try {
+    if (localStorage.getItem(noticeKey)) {
+      return;
+    }
+    localStorage.setItem(noticeKey, '1');
+  } catch (e) {
+    // If storage is blocked, just skip the one-time guard.
+  }
+
+  const show = () => {
+    if (typeof showToast === 'function') {
+      showToast(noticeText, 'error');
+    }
+  };
+
+  if (document.readyState === 'loading') {
+    document.addEventListener('DOMContentLoaded', show);
+  } else {
+    show();
+  }
+})();

docs/README.en.md

@@ -5,6 +5,9 @@
 > [!NOTE]
 > This project is for learning and research only. You must comply with Grok **Terms of Use** and **local laws and regulations**. Do not use for illegal purposes.
 
+> [!NOTE]
+> Open source projects welcome everyone's support for secondary development and pull requests, but please retain the original author's and frontend's logos to respect the work of others!
+
 Grok2API rebuilt with **FastAPI**, fully aligned with the latest web call format. Supports streaming/non-streaming chat, image generation/editing, video generation/upscale (text-to-video and image-to-video), deep reasoning, token pool concurrency, and automatic load balancing.
 
 <img width="2618" height="1658" alt="image" src="https://github.com/user-attachments/assets/a8c406f8-4c28-483a-8099-c23df5df7605" />

readme.md

@@ -5,6 +5,9 @@
 > [!NOTE]
 > 本项目仅供学习与研究，使用者必须在遵循 Grok 的 **使用条款** 以及 **法律法规** 的情况下使用，不得用于非法用途。
 
+> [!NOTE]
+> 开源项目欢迎大家支持二开和PR，但请保留原作者标识和前端标识，尊重他人劳动成果～！
+
 基于 **FastAPI** 重构的 Grok2API，全面适配最新 Web 调用格式，支持流/非流式对话、图像生成/编辑、视频生成/超分（文生视频 / 图生视频）、深度思考，号池并发与自动负载均衡一体化。
 
 <img width="2618" height="1658" alt="image" src="https://github.com/user-attachments/assets/a8c406f8-4c28-483a-8099-c23df5df7605" />

scripts/test_usage_response.py

@@ -0,0 +1,62 @@
+"""
+Simple usage probe to inspect /rest/rate-limits response.
+
+Usage:
+  python scripts/test_usage_response.py
+  (optional) TOKEN_POOL=ssoBasic|ssoSuper
+"""
+
+import asyncio
+import os
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from curl_cffi.requests import AsyncSession
+
+from app.core.config import config
+from app.services.reverse.rate_limits import RateLimitsReverse
+from app.services.token import get_token_manager
+
+
+async def main() -> int:
+    await config.load()
+    token = None
+    pool = os.getenv("TOKEN_POOL")
+    manager = await get_token_manager()
+    await manager.reload_if_stale()
+
+    if pool:
+        token = manager.get_token(pool_name=pool)
+    else:
+        token = manager.get_token(pool_name="ssoBasic") or manager.get_token(
+            pool_name="ssoSuper"
+        )
+
+    if not token:
+        token = os.getenv("GROK_TOKEN") or os.getenv("SSO_TOKEN") or os.getenv("TOKEN")
+    if not token:
+        print("Missing token. Ensure token pool is configured or set GROK_TOKEN.")
+        return 2
+
+    async with AsyncSession() as session:
+        response = await RateLimitsReverse.request(session, token)
+
+    try:
+        data = response.json()
+    except Exception as exc:
+        print(f"Failed to parse JSON: {exc}")
+        raw = getattr(response, "text", "")
+        if raw:
+            print(raw)
+        return 3
+
+    print(data)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))