# Writing a connector

A connector is a Python class under `src/aiatlas/connectors/<group>/<name>.py`, listed in that module's `CONNECTORS = [Cls]`.
The registry auto-discovers it; `aia seed` registers it in the database; `aia run <name>` runs it; the scheduler runs it on its interval.

## Skeleton

```python
from aiatlas.registry import org_ref, provider_ref
from aiatlas.sdk.connector import BaseConnector, Parsed, RunContext
from aiatlas.sdk.facts import Facts, Target
from aiatlas.sdk.fetch import FetchResult

class ExampleConnector(BaseConnector):
    name = "example"                       # unique, snake_case, stable (used in documents/claims/events)
    label = "Example Lab — models & news"
    description = "Official docs and blog of Example Lab."
    source_key = "example.com"             # must exist in registry/sources.yaml (tier, rate limit, org)
    version = "1"; parser_version = "1"    # bump parser_version when extraction improves → `aia reprocess example`
    interval_seconds = 3600; min_interval_seconds = 1800; max_interval_seconds = 86400
    rate_per_min = 15; tier = 1; priority = 0 | 1 | 2
    expected_min_records = 5               # breakage detection: fewer entities/prices/results ⇒ run "suspect", nothing deleted
    concurrency = 2

    async def discover(self, ctx: RunContext) -> list[Target]:
        return [Target(url="https://example.com/models", doc_type="model_docs", key="models", min_bytes=2000),
                Target(url="https://example.com/blog/rss.xml", doc_type="feed", key="feed")]

    async def extract(self, ctx: RunContext, target: Target, res: FetchResult, parsed: Parsed) -> Facts:
        facts = Facts()
        org = org_ref("example")            # registry organization → deterministic identifiers
        facts.entities.append(org)
        if target.key == "models" and parsed.html:
            for row in parsed.html.tables[0]["rows"]: ...
                model = facts.entity("model", name, identifiers={"example_model_id": api_id}, organization=org)
                facts.relate(org, "develops", model)
                facts.claim(model, "context_length", 128000, unit="tokens")
                facts.price(model=model, provider=provider_ref("example"), input_per_mtok=1.0, output_per_mtok=4.0)
                facts.follow(detail_url, doc_type="model_page", entity=model, needs_llm=True, meta={"llm_task": "model_passport"})
        elif target.key == "feed" and parsed.kind == "feed":
            announcement_events(facts, org, parsed.feed_items, source_name="example.com/blog")   # from connectors/labs/_common.py
        return facts

CONNECTORS = [ExampleConnector]
```

## What `Parsed` gives you

| `parsed.kind` | fields |
|---|---|
| `html` | `parsed.html: HtmlDoc` — `title, canonical, description, meta, og, json_ld, embedded_json (__NEXT_DATA__, data-props…), headings, tables [{headers, rows}], links [(href, text)], text`, `css()`, `links_matching()` ; `find_in_json(obj, key)` helper |
| `markdown` | `parsed.markdown: MarkdownDoc` — `front_matter, headings, tables, links, text, section(pattern)` |
| `feed` | `parsed.feed_items: list[FeedItem]` (`url, title, summary, published_at, categories, authors`) |
| `json` | `parsed.json` |
| `pdf` / `text` / `xml` | `parsed.text` |

Helpers: `aiatlas.sdk.extract.numbers` (`parse_param_count("70B")`, `parse_active_params("235B-A22B")`, `parse_context_length("128K")`,
`parse_money_per_mtok("$3 / 1M tokens")`, `parse_percent`), `aiatlas.sdk.extract.dates` (`parse_datetime`, `parse_date` with precision),
`connectors/labs/_common.py` (`announcement_events`, `transpose_feature_table`, `kv_tables`, `clean_cell`, `money`, `tokens`, `month_year`, `parse_retirement`, `model_ref`).

## Rules

1. **Direct mode first.** Use official pages, feeds, sitemaps, Markdown/raw files, embedded JSON. Public JSON files (e.g. `openrouter.ai/api/v1/models`,
   `pypi.org/pypi/<pkg>/json`, arXiv Atom) are fine — they are public documents, not commercial APIs. Never require an API key.
   Set `Target(escalate=True)` only for pages known to block bots *and* only if a key is configured; otherwise let the document be `blocked`.
2. **Never fabricate.** Only emit a claim when the page states it. Unknown → no claim. Don't infer parameter counts from names unless the
   name literally contains them (`Qwen3-235B-A22B` → 235e9 / 22e9 is fine; "Large" is not).
3. **Identifiers make resolution deterministic.** Give models the provider's API id (`{provider}_model_id`), HF repo (`hf_repo`), arXiv id
   (`arxiv`), GitHub repo (`github_repo`), PyPI name (`pypi`). Organizations always come from `org_ref(<registry key>)` (add missing orgs to
   `registry/organizations.yaml` with a `source_url`).
4. **Properties are shared vocabulary** (see below). Add new ones sparingly; prefix metrics with `metric.` (downloads, likes, stars) so they
   never generate events.
5. **Events**: NEW_*, PRICE_CHANGED, CONTEXT_CHANGED… are emitted automatically by the writer. Emit `ANNOUNCEMENT`/`RELEASE` events yourself for
   feed items (`announcement_events`) with `effective_at` = publication date and a `dedupe_key` (URL).
6. **Follow-ups** (`facts.follow`) let a listing discover detail pages; keep `max_targets` reasonable (`ctx.max_targets`, default 2000).
7. **Every connector has a fixture test**: save real responses under `tests/fixtures/<connector>/…` and assert extracted facts
   (see `tests/test_anthropic.py`). Run connectors with `--file key=path` to use fixtures instead of the network.
8. Respect `rate_per_min` from `registry/sources.yaml`; arXiv ≤ 4/min; Hugging Face ≤ 30/min; GitHub ≤ 20/min.

## Property vocabulary (entities.attributes)

**model**: `family, version, release_date (ISO, may be YYYY-MM), status (active|preview|deprecated|retired|announced|limited-availability), openness
(open-weights|open-source|proprietary|restricted), license, architecture, parameter_count (int), active_parameter_count, is_moe, modalities
(list of text|image|audio|video|code|embedding), modalities_input, modalities_output, context_length (tokens), max_output_tokens, knowledge_cutoff
(YYYY-MM), training_data_cutoff, languages, tool_calling, structured_output, reasoning, vision, audio, fine_tuning_available, tokenizer,
api_model_id, api_alias, official_url, model_card_url, paper_url, repository_url, hf_repo, pipeline_tag, base_model, quantization, quant_format
(gguf|mlx|awq|gptq|fp8), file_size_gb, deprecation_date, retirement_date, retirement_tentative, metric.downloads, metric.likes`

**company / organization / lab**: `country (ISO-2), headquarters, founded, website, domains, hf_org, github_org, org_kind, legal_name, founders, leadership, employee_count`

**provider**: `website, pricing_url, docs_url, regions, features` · **paper**: `authors, published_at, updated_at, abstract, arxiv_id, doi, categories, primary_category, pdf_url, code_url, venue`
· **benchmark**: `category, task, metric, unit, creator, website, paper, known_limitations, methodology` · **hardware**: `kind, architecture, release_date, memory_gb,
memory_type, memory_bandwidth_gbs, tdp_watts, runtimes, manufacturer, spec_url, price_usd, compute_fp16_tflops` · **framework / repository**: `repository_url, latest_version,
latest_release_at, license, language, description, topics, pypi, metric.stars, metric.forks` · **dataset**: `license, modality, size, publisher, task, hf_repo`

Relations: `develops, owns, operates, available_through, evaluated_on, described_by, derived_from, fine_tuned_from, quantized_from, distilled_from,
merged_from, superseded_by, runs_on, uses, manufactures, funded_by, acquired, authored, works_at, uses_dataset, evaluates_on, integrates, published_by`.