HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""xAI — developer docs models page (Next.js RSC payload embeds `globalThis.__XAI_PUBLIC_MODELS__`) + per-model Markdown pages.23Sources (tier 1):4 * models → docs.x.ai/developers/models (the /docs/models URL redirects here). The RSC payload carries a JSON catalog per region cluster:5 languageModels (name, aliases, modalities, maxPromptLength, prices in 1e-4 USD per 1M tokens, long-context threshold, features),6 imageGenerationModels / videoGenerationModels (per-image / per-second prices in 1e-10 USD), audioModels.7 * model pages → docs.x.ai/developers/models/<id>.md : "At a glance" (context window, model name, batch), capabilities, knowledge cut-off8 * x.ai/news answers 403 to crawlers → targeted with `escalate=True` only; without an escalation transport it is recorded as `blocked`.910Price units are verified against the human-readable table of models.md (grok-4.6: 20000 → $2.00 / 1M input tokens).11"""12from __future__ import annotations1314import re15from typing import Any1617from aiatlas.registry import org_ref, provider_ref18from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext19from aiatlas.sdk.extract.dates import parse_datetime20from aiatlas.sdk.facts import EntityRef, Facts, Target21from aiatlas.sdk.fetch import FetchResult2223from ._common import json_after, model_ref, next_flight_payload, tokens2425DOCS = "https://docs.x.ai/developers"26NEWS = "https://x.ai/news"27PROVIDER_KEY = "xai"28ID_SCHEME = "xai_model_id"29TOKEN_PRICE_DIVISOR = 10_000 # "20000" → 2.0 USD per 1M tokens30UNIT_PRICE_DIVISOR = 10_000_000_000 # "400000000" → 0.04 USD per image / per second31MAX_MODEL_PAGES = 2032MODALITY = {"TEXT": "text", "IMAGE": "image", "AUDIO": "audio", "VIDEO": "video"}333435class XAIConnector(BaseConnector):36 name = "xai"37 label = "xAI — Grok models & pricing"38 description = "xAI developer docs: public model catalog embedded in the models page (ids, aliases, modalities, context, prices, features) and model pages."39 source_key = "x.ai"40 version = "1"41 parser_version = "1"42 interval_seconds = 720043 min_interval_seconds = 360044 rate_per_min = 1045 tier = 146 priority = 147 expected_min_records = 1048 concurrency = 24950 async def discover(self, ctx: RunContext) -> list[Target]:51 return [52 Target(url=f"{DOCS}/models", doc_type="model_docs", key="models", min_bytes=20000),53 Target(url=NEWS, doc_type="listing", key="news", min_bytes=2000, escalate=True, priority=3),54 ]5556 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:57 facts = Facts()58 org = org_ref("xai")59 facts.entities.append(org)60 key = target.key or ""61 if key == "models" and parsed.html:62 self._models(facts, org, res)63 elif target.doc_type == "model_page" and parsed.markdown:64 self._model_page(facts, org, target, parsed)65 elif key == "news" and parsed.html:66 from aiatlas.sdk.extract.feeds import FeedItem6768 from ._common import announcement_events6970 items = [FeedItem(id=h, url=h, title=t, summary=None, published_at=None, updated_at=None)71 for h, t in parsed.html.links_matching(r"^https://x\.ai/news/[^/?#]+$") if len(t) > 12]72 announcement_events(facts, org, items, source_name="x.ai/news", max_follow=10)73 facts.document_title, facts.document_entity = "xAI news", org74 return facts7576 # ------------------------------------------------------------------------------------------ models catalog (embedded JSON)77 def _models(self, facts: Facts, org: EntityRef, res: FetchResult) -> None:78 payload = next_flight_payload(res.content)79 data = json_after(payload, "__XAI_PUBLIC_MODELS__=") if payload else None80 if not isinstance(data, dict):81 data = json_after(res.text.replace('\\"', '"'), "__XAI_PUBLIC_MODELS__=")82 if not isinstance(data, dict):83 return84 provider = provider_ref(PROVIDER_KEY)85 facts.entities.append(provider)86 facts.document_title, facts.document_entity = "xAI models", provider87 merged: dict[str, dict[str, Any]] = {}88 for cluster in data.get("clusterConfigs") or []:89 region = cluster.get("clusterName")90 for kind in ("languageModels", "imageGenerationModels", "videoGenerationModels", "audioModels"):91 for m in cluster.get(kind) or []:92 name_ = m.get("name")93 if not name_:94 continue95 entry = merged.setdefault(name_, {"kind": kind, "model": m, "regions": []})96 if region and region not in entry["regions"]:97 entry["regions"].append(region)98 if len(m) > len(entry["model"]):99 entry["model"] = m100 followed = 0101 for name_, entry in merged.items():102 m, kind = entry["model"], entry["kind"]103 ref = model_ref(facts, _display(name_), org, api_id=name_, provider_key=PROVIDER_KEY, family="Grok" if name_.startswith("grok") else None,104 aliases=[name_] + [a for a in (m.get("aliases") or []) if isinstance(a, str)])105 facts.claim(ref, "api_model_id", name_)106 facts.claim(ref, "official_url", f"{DOCS}/models/{name_}")107 facts.claim(ref, "regions", sorted(entry["regions"]))108 mods_in = [MODALITY[x] for x in m.get("inputModalities") or [] if x in MODALITY]109 mods_out = [MODALITY[x] for x in m.get("outputModalities") or [] if x in MODALITY]110 facts.claim(ref, "modalities_input", mods_in or None)111 facts.claim(ref, "modalities_output", mods_out or None)112 facts.claim(ref, "modalities", sorted(set(mods_in) | set(mods_out)) or None)113 if mods_in:114 facts.claim(ref, "vision", "image" in mods_in)115 feats = m.get("features") or {}116 if kind == "languageModels":117 facts.claim(ref, "context_length", m.get("maxPromptLength") if isinstance(m.get("maxPromptLength"), int) else None, unit="tokens")118 if "functionCalling" in feats:119 facts.claim(ref, "tool_calling", bool(feats["functionCalling"]))120 if "structuredOutputs" in feats:121 facts.claim(ref, "structured_output", bool(feats["structuredOutputs"]))122 if "reasoning" in feats:123 facts.claim(ref, "reasoning", bool(feats["reasoning"]))124 efforts = (feats.get("reasoningEffortOptions") or {}).get("supportedEfforts")125 facts.claim(ref, "reasoning_effort_options", efforts if isinstance(efforts, list) else None)126 self._language_price(facts, ref, provider, name_, m, entry["regions"])127 if followed < MAX_MODEL_PAGES:128 followed += 1129 facts.follow(f"{DOCS}/models/{name_}.md", doc_type="model_page", key=f"model:{name_}", meta={"api_id": name_, "content_type": "text/markdown"},130 min_bytes=200, priority=1)131 elif kind == "imageGenerationModels":132 per_image = [(_unit(r.get("pricePerImage")), r.get("resolution"), r.get("quality")) for r in m.get("resolutionPricing") or []]133 per_image = [p for p in per_image if p[0] is not None]134 base = min((p[0] for p in per_image), default=_unit(m.get("imagePrice")))135 if base is not None:136 facts.price(model=ref, provider=provider, provider_model_id=name_, per_image=base,137 features={"per_image_by_resolution": [{"resolution": r, "quality": q, "usd": p} for p, r, q in per_image],138 "per_input_image": _unit(m.get("pricePerInputImage")), "regions": sorted(entry["regions"])}, meta={"from": "models page JSON"})139 elif kind == "videoGenerationModels":140 per_sec = {str(r.get("resolution")).replace("VIDEO_RESOLUTION_", "").lower(): _unit(r.get("pricePerSecond")) for r in m.get("resolutionPricing") or []}141 facts.claim(ref, "pricing_per_second_usd", {k: v for k, v in per_sec.items() if v is not None} or None)142 if m.get("version") and str(m["version"]) != "1.0":143 facts.claim(ref, "version", str(m["version"]))144145 def _language_price(self, facts: Facts, ref: EntityRef, provider: EntityRef, name_: str, m: dict[str, Any], regions: list[str]) -> None:146 p_in, p_cached, p_out = _tok(m.get("promptTextTokenPrice")), _tok(m.get("cachedPromptTokenPrice")), _tok(m.get("completionTextTokenPrice"))147 if p_in is None and p_out is None:148 return149 features: dict[str, Any] = {"regions": sorted(regions)}150 threshold = m.get("longContextThreshold")151 if threshold and str(threshold).isdigit():152 features["long_context_threshold_tokens"] = int(threshold)153 for key, prop in (("promptTextTokenPriceLongContext", "long_context_input_per_mtok"), ("cachedPromptTokenPriceLongContext", "long_context_cached_input_per_mtok"),154 ("completionTokenPriceLongContext", "long_context_output_per_mtok")):155 v = _tok(m.get(key))156 if v is not None:157 features[prop] = v158 img = _tok(m.get("promptImageTokenPrice"))159 if img is not None and img != p_in:160 features["image_input_per_mtok"] = img161 if m.get("batchDiscountPercent"):162 features["batch_discount_percent"] = m["batchDiscountPercent"]163 if p_in is not None:164 pass165 features["batch_enabled"] = bool(m.get("batchEnabled"))166 batch_in = batch_out = None167 if m.get("batchEnabled") and m.get("batchDiscountPercent"):168 factor = 1 - float(m["batchDiscountPercent"]) / 100169 batch_in = round(p_in * factor, 6) if p_in is not None else None170 batch_out = round(p_out * factor, 6) if p_out is not None else None171 facts.price(model=ref, provider=provider, provider_model_id=name_, input_per_mtok=p_in, cached_input_per_mtok=p_cached, output_per_mtok=p_out,172 batch_input_per_mtok=batch_in, batch_output_per_mtok=batch_out, context_length=m.get("maxPromptLength") if isinstance(m.get("maxPromptLength"), int) else None,173 features=features, meta={"from": "models page JSON"})174175 # ------------------------------------------------------------------------------------------ model page (.md)176 def _model_page(self, facts: Facts, org: EntityRef, target: Target, parsed: Parsed) -> None:177 md = parsed.markdown178 assert md179 api_id = target.meta.get("api_id")180 m = re.search(r"\*\*Model name:\*\*\s*`([^`]+)`", md.body)181 if m:182 api_id = m.group(1).strip()183 title = next((t for lvl, t in md.headings if lvl == 1), None)184 if not api_id or not title:185 return186 ref = model_ref(facts, title, org, api_id=api_id, provider_key=PROVIDER_KEY, aliases=[api_id])187 facts.document_title, facts.document_entity = title, ref188 first_para = next((ln.strip() for ln in md.body.split("\n")[1:] if ln.strip() and not ln.startswith("#")), None)189 facts.claim(ref, "description", first_para)190 m = re.search(r"\*\*Context window:\*\*\s*([\d,]+)", md.body)191 if m:192 facts.claim(ref, "context_length", tokens(m.group(1)), unit="tokens")193 m = re.search(r"\*\*Modalities:\*\*\s*([^→\n]+)→\s*([^\n]+)", md.body)194 if m:195 mods_in = [x.strip() for x in m.group(1).split(",") if x.strip() in MODALITY.values()]196 mods_out = [x.strip() for x in m.group(2).split(",") if x.strip() in MODALITY.values()]197 facts.claim(ref, "modalities_input", mods_in or None)198 facts.claim(ref, "modalities_output", mods_out or None)199 for label, prop in (("Function calling", "tool_calling"), ("Structured outputs", "structured_output"), ("Reasoning", "reasoning")):200 mm = re.search(rf"\*\*{label}:\*\*\s*(Yes|No)", md.body)201 if mm:202 facts.claim(ref, prop, mm.group(1) == "Yes")203 mm = re.search(r"knowledge cut-?off date of .*? is ([A-Z][a-z]+ \d{1,2}, \d{4})", md.body)204 if mm:205 dt = parse_datetime(mm.group(1))206 facts.claim(ref, "knowledge_cutoff", f"{dt.year:04d}-{dt.month:02d}" if dt else None)207 mm = re.search(r"\*\*Batch API:\*\*\s*(Supported|Not supported)", md.body)208 if mm:209 facts.claim(ref, "batch_api", mm.group(1) == "Supported")210211212# ---------------------------------------------------------------------------------------------- helpers213def _tok(v: Any) -> float | None:214 try:215 return round(int(str(v)) / TOKEN_PRICE_DIVISOR, 6) if v not in (None, "") else None216 except ValueError:217 return None218219220def _unit(v: Any) -> float | None:221 try:222 return round(int(str(v)) / UNIT_PRICE_DIVISOR, 6) if v not in (None, "") else None223 except ValueError:224 return None225226227def _display(api_id: str) -> str:228 """grok-4.6 → Grok 4.6 ; grok-imagine-image-2.0 → Grok Imagine Image 2.0 ; dated / variant ids are kept verbatim."""229 parts = api_id.split("-")230 if all(re.fullmatch(r"[a-z]+|\d+(?:\.\d+)?", p) for p in parts) and sum(bool(re.fullmatch(r"\d+(?:\.\d+)?", p)) for p in parts) <= 1:231 return " ".join((p.upper() if p in ("tts", "stt") else p.capitalize()) if p.isalpha() else p for p in parts)232 return api_id233234235CONNECTORS = [XAIConnector]236