"""Anthropic — official docs (served as Markdown by the docs platform) + newsroom listing. Sources (tier 1): * models overview → comparison table transposed into one model per column: API ids, context, max output, cutoffs, retirement * pricing → per-model price table → PriceObs for the Anthropic API provider (+ status hints such as "retired") * model deprecations → status table keyed by API model name → status / deprecation / retirement claims * model pages → per-model spec tables (discovered from the overview) * newsroom → ANNOUNCEMENT events, release articles queued for LLM extraction """ from __future__ import annotations import re from aiatlas.registry import org_ref, provider_ref from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext from aiatlas.sdk.extract.dates import parse_datetime from aiatlas.sdk.extract.feeds import FeedItem from aiatlas.sdk.facts import Facts, Target from aiatlas.sdk.fetch import FetchResult from ._common import ( announcement_events, claim_api_aliases, claim_modalities, claim_status, clean_cell, kv_tables, link_in_cell, model_ref, money, month_year, normalize_capabilities, parse_retirement, tokens, transpose_feature_table, ) DOCS = "https://docs.claude.com/en/docs/about-claude" NEWS = "https://www.anthropic.com/news" PROVIDER_KEY = "anthropic" CLAUDE_NAME = re.compile(r"^(Claude [A-Za-z]+(?: [0-9.]+)?(?: [A-Za-z]+)?)") class AnthropicConnector(BaseConnector): name = "anthropic" label = "Anthropic — models, pricing, deprecations, news" description = "Official Claude docs (models overview, pricing, deprecations, model pages) and the Anthropic newsroom." source_key = "docs.claude.com" version = "1" parser_version = "1" interval_seconds = 3600 min_interval_seconds = 1800 rate_per_min = 15 tier = 1 priority = 0 expected_min_records = 4 concurrency = 2 async def discover(self, ctx: RunContext) -> list[Target]: return [ Target(url=f"{DOCS}/models/overview.md", doc_type="model_docs", key="models", min_bytes=2000), Target(url=f"{DOCS}/pricing.md", doc_type="pricing", key="pricing", min_bytes=2000), Target(url=f"{DOCS}/model-deprecations.md", doc_type="model_docs", key="deprecations", min_bytes=1000), Target(url=NEWS, doc_type="listing", key="news", min_bytes=5000), ] async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts: facts = Facts() org = org_ref("anthropic") facts.entities.append(org) key = target.key or target.meta.get("kind") if key == "models" and parsed.markdown: self._models_overview(facts, org, parsed) elif key == "pricing" and parsed.markdown: self._pricing(facts, org, parsed) elif key == "deprecations" and parsed.markdown: self._deprecations(facts, org, parsed) elif key == "news" and parsed.html: self._news(facts, org, parsed, res) elif target.doc_type == "model_page" and parsed.markdown: self._model_page(facts, org, target, parsed) return facts # ------------------------------------------------------------------------------------------ models overview def _models_overview(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def] md = parsed.markdown assert md facts.document_title = md.front_matter.get("title") or "Models overview" table = next((t for t in md.tables if t["headers"] and clean_cell(t["headers"][0]).lower() == "feature"), None) if not table: return raw_headers = table["headers"] per_model = transpose_feature_table(table) for i, (model_name, feats) in enumerate(per_model.items()): api_id = feats.get("claude api id") or feats.get("anthropic api id") alias = feats.get("claude api alias") or feats.get("anthropic api alias") aliases = [a for a in (api_id, alias) if a] ref = model_ref(facts, model_name, org, api_id=api_id, provider_key=PROVIDER_KEY, family="Claude", aliases=aliases + _name_variants(model_name)) facts.claim(ref, "openness", "proprietary") claim_status(facts, ref, "active") facts.claim(ref, "description", feats.get("description")) facts.claim(ref, "api_model_id", api_id) claim_api_aliases(facts, ref, alias) facts.claim(ref, "context_length", tokens(feats.get("context window", "")), unit="tokens") facts.claim(ref, "max_output_tokens", tokens(feats.get("max output", "")), unit="tokens") facts.claim(ref, "knowledge_cutoff", month_year(feats.get("reliable knowledge cutoff", ""))) facts.claim(ref, "training_data_cutoff", month_year(feats.get("training data cutoff", ""))) facts.claim(ref, "latency_tier", feats.get("comparative latency")) facts.claim(ref, "thinking", feats.get("thinking") or feats.get("extended thinking")) facts.claim(ref, "default_effort", feats.get("default effort")) # modalities / capabilities only when the comparison table has a row for them (no hard-coded constants) self._table_capabilities(facts, ref, feats) retire, tentative = parse_retirement(feats.get("retirement", "")) if retire: facts.claim(ref, "retirement_date", retire) facts.claim(ref, "retirement_tentative", tentative) for platform, scheme in (("amazon bedrock id", "bedrock_model_id"), ("google cloud id", "vertex_model_id"), ("microsoft foundry id", "foundry_model_id"), ("claude platform on aws id", "claude_aws_model_id")): v = feats.get(platform) if v and v not in ("—", "-"): facts.claim(ref, scheme, v) 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) if not page_url: # header cells are plain names; the "Model page" row holds the links in the raw table for row in parsed.markdown.tables[0]["rows"] if parsed.markdown else []: pass if page_url: if not page_url.endswith(".md"): page_url = page_url.rstrip("/") + ".md" 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) facts.document_entity = org # raw table rows keep the links: recover "Model page" links precisely for t in md.tables: if t["headers"] and clean_cell(t["headers"][0]).lower() == "feature": for row in t["rows"]: if row and clean_cell(row[0]).lower() == "model page": for name, cell in zip([clean_cell(h) for h in t["headers"][1:]], row[1:], strict=False): url = link_in_cell(cell) if url and not any(f.url.startswith(url.rstrip("/")) for f in facts.targets): ref = next((e for e in facts.entities if e.entity_type == "model" and e.name == name), None) facts.follow(url.rstrip("/") + ".md", doc_type="model_page", entity=ref, key=f"model:{name}", meta={"model": name}, min_bytes=500) @staticmethod def _table_capabilities(facts: Facts, ref, feats: dict[str, str]) -> None: # type: ignore[no-untyped-def] """Rows of the comparison table that state modalities or capabilities → canonical claims; absent rows → no claim.""" caps: list[str] = [] yes = re.compile(r"^(yes|supported|✓|✔|adaptive|adaptive \(always on\)|extended|available)", re.IGNORECASE) no = re.compile(r"^(no|not supported|—|-|n/a|unsupported)$", re.IGNORECASE) for key, cap in (("thinking", "reasoning"), ("extended thinking", "reasoning"), ("vision", "vision"), ("image input", "vision"), ("tool use", "function_calling"), ("function calling", "function_calling"), ("structured outputs", "structured_output"), ("prompt caching", "caching"), ("batch api", "batch"), ("computer use", "computer_use"), ("pdf support", "document_input"), ("citations", "citations"), ("web search", "search_grounding")): v = feats.get(key) if v is None or v == "": continue if no.match(v.strip()): continue if yes.match(v.strip()) or key in ("thinking", "extended thinking"): caps.append(cap) if caps: facts.claim(ref, "capabilities", normalize_capabilities(caps)) if "function_calling" in caps: facts.claim(ref, "tool_calling", True) if "vision" in caps: facts.claim(ref, "vision", True) if "reasoning" in caps: facts.claim(ref, "reasoning", True) mods = feats.get("modalities") or feats.get("input modalities") or feats.get("input") outs = feats.get("output modalities") or feats.get("output") if mods or outs: claim_modalities(facts, ref, mods, outs) # ------------------------------------------------------------------------------------------ model page def _model_page(self, facts: Facts, org, target: Target, parsed: Parsed) -> None: # type: ignore[no-untyped-def] md = parsed.markdown assert md name = target.meta.get("model") or md.front_matter.get("title") or "" if not name: return ref = target.entity or model_ref(facts, name, org, api_id=target.meta.get("api_id"), provider_key=PROVIDER_KEY, family="Claude") if ref not in facts.entities: facts.entities.append(ref) kv = kv_tables(md.tables) facts.claim(ref, "official_url", md.front_matter.get("url")) facts.claim(ref, "description", md.front_matter.get("description")) for k, prop in (("context window", "context_length"), ("max output", "max_output_tokens")): if kv.get(k): facts.claim(ref, prop, tokens(kv[k]), unit="tokens") for k, prop in (("reliable knowledge cutoff", "knowledge_cutoff"), ("training data cutoff", "training_data_cutoff")): if kv.get(k): facts.claim(ref, prop, month_year(kv[k])) if kv.get("release date") or kv.get("released"): dt = parse_datetime(kv.get("release date") or kv.get("released")) if dt: facts.claim(ref, "release_date", dt.date().isoformat()) facts.document_entity = ref facts.document_title = md.front_matter.get("title") # ------------------------------------------------------------------------------------------ pricing def _pricing(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def] md = parsed.markdown assert md facts.document_title = md.front_matter.get("title") or "Pricing" provider = provider_ref(PROVIDER_KEY) facts.entities.append(provider) 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) if not table: return headers = [clean_cell(h).lower() for h in table["headers"]] def col(*needles: str) -> int | None: for i, h in enumerate(headers): if all(n in h for n in needles): return i return None c_in, c_out = col("input"), col("output") c_w5, c_w1, c_hit = col("5m", "cache"), col("1h", "cache"), col("cache hit") for row in table["rows"]: if not row or c_in is None or c_out is None: continue raw_name = row[0] name_clean = clean_cell(raw_name) m = CLAUDE_NAME.match(name_clean) model_name = m.group(1).strip() if m else name_clean.split("(")[0].strip() note = name_clean[len(model_name):].strip(" ()") status = None low = note.lower() if "retired" in low: status = "retired" elif "deprecated" in low: status = "deprecated" elif "limited availability" in low: status = "limited-availability" ref = model_ref(facts, model_name, org, family="Claude", aliases=_name_variants(model_name)) facts.claim(ref, "openness", "proprietary") if status: claim_status(facts, ref, status) facts.claim(ref, "availability_note", note) facts.price(model=ref, provider=provider, input_per_mtok=money(row[c_in]), output_per_mtok=money(row[c_out]), cached_input_per_mtok=money(row[c_hit]) if c_hit is not None and c_hit < len(row) else None, cache_write_per_mtok=money(row[c_w5]) if c_w5 is not None and c_w5 < len(row) else None, features={"cache_write_1h_per_mtok": money(row[c_w1]) if c_w1 is not None and c_w1 < len(row) else None}, meta={"from": "pricing page", "note": note or None}) # batch discount, if stated as a multiplier/percent for t in md.tables: hs = [clean_cell(h).lower() for h in t["headers"]] if hs and "cache operation" in hs[0]: facts.claim(provider, "prompt_caching", {clean_cell(r[0]): clean_cell(r[1]) for r in t["rows"] if len(r) >= 2}) facts.document_entity = provider # ------------------------------------------------------------------------------------------ deprecations def _deprecations(self, facts: Facts, org, parsed: Parsed) -> None: # type: ignore[no-untyped-def] md = parsed.markdown assert md facts.document_title = md.front_matter.get("title") or "Model deprecations" table = next((t for t in md.tables if t["headers"] and "api model name" in clean_cell(t["headers"][0]).lower()), None) if not table: return headers = [clean_cell(h).lower() for h in table["headers"]] i_state = next((i for i, h in enumerate(headers) if "state" in h or "status" in h), 1) i_dep = next((i for i, h in enumerate(headers) if "deprecated" in h), 2) i_ret = next((i for i, h in enumerate(headers) if "retirement" in h), 3) for row in table["rows"]: if len(row) <= max(i_state, i_dep, i_ret): continue api_id = clean_cell(row[0]) if not api_id.startswith("claude"): continue display = _display_name(api_id) ref = model_ref(facts, display, org, api_id=api_id, provider_key=PROVIDER_KEY, family="Claude", aliases=[api_id] + _name_variants(display)) state = clean_cell(row[i_state]).lower() claim_status(facts, ref, state) dep = clean_cell(row[i_dep]) if dep and dep.upper() != "N/A": d = parse_datetime(dep) if d: facts.claim(ref, "deprecation_date", d.date().isoformat()) ret, tentative = parse_retirement(clean_cell(row[i_ret])) if ret: facts.claim(ref, "retirement_date", ret) facts.claim(ref, "retirement_tentative", tentative) facts.document_entity = org # ------------------------------------------------------------------------------------------ news def _news(self, facts: Facts, org, parsed: Parsed, res: FetchResult) -> None: # type: ignore[no-untyped-def] html = parsed.html assert html items: list[FeedItem] = [] seen: set[str] = set() for node in html.css("a[href^='/news/']"): href = node.attributes.get("href") or "" if href in seen or href.count("/") != 2: continue text = re.sub(r"\s+", " ", node.text(separator=" | ", strip=True)) parts = [p.strip() for p in text.split("|") if p.strip()] if len(parts) < 2: continue date = None category = None title = None summary = None for p in parts: if date is None and re.fullmatch(r"[A-Z][a-z]{2} \d{1,2}, \d{4}", p): date = parse_datetime(p) elif category is None and p in ("Announcements", "Product", "Policy", "Research", "Interpretability", "Alignment", "Societal Impacts", "Economic Research", "Education"): category = p elif title is None: title = p elif summary is None: summary = p if not title: continue seen.add(href) items.append(FeedItem(id=href, url=f"https://www.anthropic.com{href}", title=title[:300], summary=summary, published_at=date, updated_at=None, categories=[category] if category else [])) announcement_events(facts, org, items, source_name="anthropic.com/news", follow=True, max_follow=25) facts.document_title = "Anthropic newsroom" facts.document_entity = org def _name_variants(name: str) -> list[str]: """Anthropic has used both 'Claude 3.5 Haiku' and 'Claude Haiku 3.5' — register both orders as aliases.""" m = re.fullmatch(r"Claude (\d[\d.]*) ([A-Za-z]+)", name) if m: return [f"Claude {m.group(2)} {m.group(1)}"] m = re.fullmatch(r"Claude ([A-Za-z]+) (\d[\d.]*)", name) if m: return [f"Claude {m.group(2)} {m.group(1)}"] return [] def _display_name(api_id: str) -> str: """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""" parts = api_id.split("-") parts = [p for p in parts if not re.fullmatch(r"\d{8}", p)] words: list[str] = [] nums: list[str] = [] for p in parts: if p.isdigit(): nums.append(p) else: if nums: words.append(".".join(nums)) nums = [] words.append(p.capitalize()) if nums: words.append(".".join(nums)) return " ".join(words) CONNECTORS = [AnthropicConnector]