HTML 77.2%
TypeScript 10.5%
Python 9.6%
JavaScript 2.5%
1"""Anthropic — official docs (served as Markdown by the docs platform) + newsroom listing.23Sources (tier 1):4 * models overview → comparison table transposed into one model per column: API ids, context, max output, cutoffs, retirement5 * pricing → per-model price table → PriceObs for the Anthropic API provider (+ status hints such as "retired")6 * model deprecations → status table keyed by API model name → status / deprecation / retirement claims7 * model pages → per-model spec tables (discovered from the overview)8 * newsroom → ANNOUNCEMENT events, release articles queued for LLM extraction9"""10from __future__ import annotations1112import re1314from aiatlas.registry import org_ref, provider_ref15from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext16from aiatlas.sdk.extract.dates import parse_datetime17from aiatlas.sdk.extract.feeds import FeedItem18from aiatlas.sdk.facts import Facts, Target19from aiatlas.sdk.fetch import FetchResult2021from ._common import (22 announcement_events,23 claim_api_aliases,24 claim_modalities,25 claim_status,26 clean_cell,27 kv_tables,28 link_in_cell,29 model_ref,30 money,31 month_year,32 normalize_capabilities,33 parse_retirement,34 tokens,35 transpose_feature_table,36)3738DOCS = "https://docs.claude.com/en/docs/about-claude"39NEWS = "https://www.anthropic.com/news"40PROVIDER_KEY = "anthropic"4142CLAUDE_NAME = re.compile(r"^(Claude [A-Za-z]+(?: [0-9.]+)?(?: [A-Za-z]+)?)")434445class AnthropicConnector(BaseConnector):46 name = "anthropic"47 label = "Anthropic — models, pricing, deprecations, news"48 description = "Official Claude docs (models overview, pricing, deprecations, model pages) and the Anthropic newsroom."49 source_key = "docs.claude.com"50 version = "1"51 parser_version = "1"52 interval_seconds = 360053 min_interval_seconds = 180054 rate_per_min = 1555 tier = 156 priority = 057 expected_min_records = 458 concurrency = 25960 async def discover(self, ctx: RunContext) -> list[Target]:61 return [62 Target(url=f"{DOCS}/models/overview.md", doc_type="model_docs", key="models", min_bytes=2000),63 Target(url=f"{DOCS}/pricing.md", doc_type="pricing", key="pricing", min_bytes=2000),64 Target(url=f"{DOCS}/model-deprecations.md", doc_type="model_docs", key="deprecations", min_bytes=1000),65 Target(url=NEWS, doc_type="listing", key="news", min_bytes=5000),66 ]6768 async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:69 facts = Facts()70 org = org_ref("anthropic")71 facts.entities.append(org)72 key = target.key or target.meta.get("kind")73 if key == "models" and parsed.markdown:74 self._models_overview(facts, org, parsed)75 elif key == "pricing" and parsed.markdown:76 self._pricing(facts, org, parsed)77 elif key == "deprecations" and parsed.markdown:78 self._deprecations(facts, org, parsed)79 elif key == "news" and parsed.html:80 self._news(facts, org, parsed, res)81 elif target.doc_type == "model_page" and parsed.markdown:82 self._model_page(facts, org, target, parsed)83 return facts8485 # ------------------------------------------------------------------------------------------ models overview86 def _models_overview(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def]87 md = parsed.markdown88 assert md89 facts.document_title = md.front_matter.get("title") or "Models overview"90 table = next((t for t in md.tables if t["headers"] and clean_cell(t["headers"][0]).lower() == "feature"), None)91 if not table:92 return93 raw_headers = table["headers"]94 per_model = transpose_feature_table(table)95 for i, (model_name, feats) in enumerate(per_model.items()):96 api_id = feats.get("claude api id") or feats.get("anthropic api id")97 alias = feats.get("claude api alias") or feats.get("anthropic api alias")98 aliases = [a for a in (api_id, alias) if a]99 ref = model_ref(facts, model_name, org, api_id=api_id, provider_key=PROVIDER_KEY, family="Claude", aliases=aliases + _name_variants(model_name))100 facts.claim(ref, "openness", "proprietary")101 claim_status(facts, ref, "active")102 facts.claim(ref, "description", feats.get("description"))103 facts.claim(ref, "api_model_id", api_id)104 claim_api_aliases(facts, ref, alias)105 facts.claim(ref, "context_length", tokens(feats.get("context window", "")), unit="tokens")106 facts.claim(ref, "max_output_tokens", tokens(feats.get("max output", "")), unit="tokens")107 facts.claim(ref, "knowledge_cutoff", month_year(feats.get("reliable knowledge cutoff", "")))108 facts.claim(ref, "training_data_cutoff", month_year(feats.get("training data cutoff", "")))109 facts.claim(ref, "latency_tier", feats.get("comparative latency"))110 facts.claim(ref, "thinking", feats.get("thinking") or feats.get("extended thinking"))111 facts.claim(ref, "default_effort", feats.get("default effort"))112 # modalities / capabilities only when the comparison table has a row for them (no hard-coded constants)113 self._table_capabilities(facts, ref, feats)114 retire, tentative = parse_retirement(feats.get("retirement", ""))115 if retire:116 facts.claim(ref, "retirement_date", retire)117 facts.claim(ref, "retirement_tentative", tentative)118 for platform, scheme in (("amazon bedrock id", "bedrock_model_id"), ("google cloud id", "vertex_model_id"), ("microsoft foundry id", "foundry_model_id"),119 ("claude platform on aws id", "claude_aws_model_id")):120 v = feats.get(platform)121 if v and v not in ("—", "-"):122 facts.claim(ref, scheme, v)123 page_url = link_in_cell(feats.get("model page", "")) or (link_in_cell(raw_headers[i + 1]) if i + 1 < len(raw_headers) else None)124 if not page_url:125 # header cells are plain names; the "Model page" row holds the links in the raw table126 for row in parsed.markdown.tables[0]["rows"] if parsed.markdown else []:127 pass128 if page_url:129 if not page_url.endswith(".md"):130 page_url = page_url.rstrip("/") + ".md"131 facts.follow(page_url, doc_type="model_page", entity=ref, key=f"model:{api_id or model_name}", meta={"model": model_name, "api_id": api_id}, min_bytes=500)132 facts.document_entity = org133 # raw table rows keep the links: recover "Model page" links precisely134 for t in md.tables:135 if t["headers"] and clean_cell(t["headers"][0]).lower() == "feature":136 for row in t["rows"]:137 if row and clean_cell(row[0]).lower() == "model page":138 for name, cell in zip([clean_cell(h) for h in t["headers"][1:]], row[1:], strict=False):139 url = link_in_cell(cell)140 if url and not any(f.url.startswith(url.rstrip("/")) for f in facts.targets):141 ref = next((e for e in facts.entities if e.entity_type == "model" and e.name == name), None)142 facts.follow(url.rstrip("/") + ".md", doc_type="model_page", entity=ref, key=f"model:{name}", meta={"model": name}, min_bytes=500)143144 @staticmethod145 def _table_capabilities(facts: Facts, ref, feats: dict[str, str]) -> None: # type: ignore[no-untyped-def]146 """Rows of the comparison table that state modalities or capabilities → canonical claims; absent rows → no claim."""147 caps: list[str] = []148 yes = re.compile(r"^(yes|supported|✓|✔|adaptive|adaptive \(always on\)|extended|available)", re.IGNORECASE)149 no = re.compile(r"^(no|not supported|—|-|n/a|unsupported)$", re.IGNORECASE)150 for key, cap in (("thinking", "reasoning"), ("extended thinking", "reasoning"), ("vision", "vision"), ("image input", "vision"), ("tool use", "function_calling"),151 ("function calling", "function_calling"), ("structured outputs", "structured_output"), ("prompt caching", "caching"), ("batch api", "batch"),152 ("computer use", "computer_use"), ("pdf support", "document_input"), ("citations", "citations"), ("web search", "search_grounding")):153 v = feats.get(key)154 if v is None or v == "":155 continue156 if no.match(v.strip()):157 continue158 if yes.match(v.strip()) or key in ("thinking", "extended thinking"):159 caps.append(cap)160 if caps:161 facts.claim(ref, "capabilities", normalize_capabilities(caps))162 if "function_calling" in caps:163 facts.claim(ref, "tool_calling", True)164 if "vision" in caps:165 facts.claim(ref, "vision", True)166 if "reasoning" in caps:167 facts.claim(ref, "reasoning", True)168 mods = feats.get("modalities") or feats.get("input modalities") or feats.get("input")169 outs = feats.get("output modalities") or feats.get("output")170 if mods or outs:171 claim_modalities(facts, ref, mods, outs)172173 # ------------------------------------------------------------------------------------------ model page174 def _model_page(self, facts: Facts, org, target: Target, parsed: Parsed) -> None: # type: ignore[no-untyped-def]175 md = parsed.markdown176 assert md177 name = target.meta.get("model") or md.front_matter.get("title") or ""178 if not name:179 return180 ref = target.entity or model_ref(facts, name, org, api_id=target.meta.get("api_id"), provider_key=PROVIDER_KEY, family="Claude")181 if ref not in facts.entities:182 facts.entities.append(ref)183 kv = kv_tables(md.tables)184 facts.claim(ref, "official_url", md.front_matter.get("url"))185 facts.claim(ref, "description", md.front_matter.get("description"))186 for k, prop in (("context window", "context_length"), ("max output", "max_output_tokens")):187 if kv.get(k):188 facts.claim(ref, prop, tokens(kv[k]), unit="tokens")189 for k, prop in (("reliable knowledge cutoff", "knowledge_cutoff"), ("training data cutoff", "training_data_cutoff")):190 if kv.get(k):191 facts.claim(ref, prop, month_year(kv[k]))192 if kv.get("release date") or kv.get("released"):193 dt = parse_datetime(kv.get("release date") or kv.get("released"))194 if dt:195 facts.claim(ref, "release_date", dt.date().isoformat())196 facts.document_entity = ref197 facts.document_title = md.front_matter.get("title")198199 # ------------------------------------------------------------------------------------------ pricing200 def _pricing(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def]201 md = parsed.markdown202 assert md203 facts.document_title = md.front_matter.get("title") or "Pricing"204 provider = provider_ref(PROVIDER_KEY)205 facts.entities.append(provider)206 table = next((t for t in md.tables if t["headers"] and clean_cell(t["headers"][0]).lower() == "model" and any("output" in clean_cell(h).lower() for h in t["headers"])), None)207 if not table:208 return209 headers = [clean_cell(h).lower() for h in table["headers"]]210211 def col(*needles: str) -> int | None:212 for i, h in enumerate(headers):213 if all(n in h for n in needles):214 return i215 return None216217 c_in, c_out = col("input"), col("output")218 c_w5, c_w1, c_hit = col("5m", "cache"), col("1h", "cache"), col("cache hit")219 for row in table["rows"]:220 if not row or c_in is None or c_out is None:221 continue222 raw_name = row[0]223 name_clean = clean_cell(raw_name)224 m = CLAUDE_NAME.match(name_clean)225 model_name = m.group(1).strip() if m else name_clean.split("(")[0].strip()226 note = name_clean[len(model_name):].strip(" ()")227 status = None228 low = note.lower()229 if "retired" in low:230 status = "retired"231 elif "deprecated" in low:232 status = "deprecated"233 elif "limited availability" in low:234 status = "limited-availability"235 ref = model_ref(facts, model_name, org, family="Claude", aliases=_name_variants(model_name))236 facts.claim(ref, "openness", "proprietary")237 if status:238 claim_status(facts, ref, status)239 facts.claim(ref, "availability_note", note)240 facts.price(model=ref, provider=provider, input_per_mtok=money(row[c_in]), output_per_mtok=money(row[c_out]),241 cached_input_per_mtok=money(row[c_hit]) if c_hit is not None and c_hit < len(row) else None,242 cache_write_per_mtok=money(row[c_w5]) if c_w5 is not None and c_w5 < len(row) else None,243 features={"cache_write_1h_per_mtok": money(row[c_w1]) if c_w1 is not None and c_w1 < len(row) else None},244 meta={"from": "pricing page", "note": note or None})245 # batch discount, if stated as a multiplier/percent246 for t in md.tables:247 hs = [clean_cell(h).lower() for h in t["headers"]]248 if hs and "cache operation" in hs[0]:249 facts.claim(provider, "prompt_caching", {clean_cell(r[0]): clean_cell(r[1]) for r in t["rows"] if len(r) >= 2})250 facts.document_entity = provider251252 # ------------------------------------------------------------------------------------------ deprecations253 def _deprecations(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def]254 md = parsed.markdown255 assert md256 facts.document_title = md.front_matter.get("title") or "Model deprecations"257 table = next((t for t in md.tables if t["headers"] and "api model name" in clean_cell(t["headers"][0]).lower()), None)258 if not table:259 return260 headers = [clean_cell(h).lower() for h in table["headers"]]261 i_state = next((i for i, h in enumerate(headers) if "state" in h or "status" in h), 1)262 i_dep = next((i for i, h in enumerate(headers) if "deprecated" in h), 2)263 i_ret = next((i for i, h in enumerate(headers) if "retirement" in h), 3)264 for row in table["rows"]:265 if len(row) <= max(i_state, i_dep, i_ret):266 continue267 api_id = clean_cell(row[0])268 if not api_id.startswith("claude"):269 continue270 display = _display_name(api_id)271 ref = model_ref(facts, display, org, api_id=api_id, provider_key=PROVIDER_KEY, family="Claude", aliases=[api_id] + _name_variants(display))272 state = clean_cell(row[i_state]).lower()273 claim_status(facts, ref, state)274 dep = clean_cell(row[i_dep])275 if dep and dep.upper() != "N/A":276 d = parse_datetime(dep)277 if d:278 facts.claim(ref, "deprecation_date", d.date().isoformat())279 ret, tentative = parse_retirement(clean_cell(row[i_ret]))280 if ret:281 facts.claim(ref, "retirement_date", ret)282 facts.claim(ref, "retirement_tentative", tentative)283 facts.document_entity = org284285 # ------------------------------------------------------------------------------------------ news286 def _news(self, facts: Facts, org, parsed: Parsed, res: FetchResult) -> None: # type: ignore[no-untyped-def]287 html = parsed.html288 assert html289 items: list[FeedItem] = []290 seen: set[str] = set()291 for node in html.css("a[href^='/news/']"):292 href = node.attributes.get("href") or ""293 if href in seen or href.count("/") != 2:294 continue295 text = re.sub(r"\s+", " ", node.text(separator=" | ", strip=True))296 parts = [p.strip() for p in text.split("|") if p.strip()]297 if len(parts) < 2:298 continue299 date = None300 category = None301 title = None302 summary = None303 for p in parts:304 if date is None and re.fullmatch(r"[A-Z][a-z]{2} \d{1,2}, \d{4}", p):305 date = parse_datetime(p)306 elif category is None and p in ("Announcements", "Product", "Policy", "Research", "Interpretability", "Alignment", "Societal Impacts", "Economic Research", "Education"):307 category = p308 elif title is None:309 title = p310 elif summary is None:311 summary = p312 if not title:313 continue314 seen.add(href)315 items.append(FeedItem(id=href, url=f"https://www.anthropic.com{href}", title=title[:300], summary=summary, published_at=date, updated_at=None,316 categories=[category] if category else []))317 announcement_events(facts, org, items, source_name="anthropic.com/news", follow=True, max_follow=25)318 facts.document_title = "Anthropic newsroom"319 facts.document_entity = org320321322def _name_variants(name: str) -> list[str]:323 """Anthropic has used both 'Claude 3.5 Haiku' and 'Claude Haiku 3.5' — register both orders as aliases."""324 m = re.fullmatch(r"Claude (\d[\d.]*) ([A-Za-z]+)", name)325 if m:326 return [f"Claude {m.group(2)} {m.group(1)}"]327 m = re.fullmatch(r"Claude ([A-Za-z]+) (\d[\d.]*)", name)328 if m:329 return [f"Claude {m.group(2)} {m.group(1)}"]330 return []331332333def _display_name(api_id: str) -> str:334 """claude-opus-4-5-20251101 → Claude Opus 4.5 ; claude-3-7-sonnet-20250219 → Claude 3.7 Sonnet ; claude-fable-5-1 → Claude Fable 5.1"""335 parts = api_id.split("-")336 parts = [p for p in parts if not re.fullmatch(r"\d{8}", p)]337 words: list[str] = []338 nums: list[str] = []339 for p in parts:340 if p.isdigit():341 nums.append(p)342 else:343 if nums:344 words.append(".".join(nums))345 nums = []346 words.append(p.capitalize())347 if nums:348 words.append(".".join(nums))349 return " ".join(words)350351352CONNECTORS = [AnthropicConnector]353