diff --git a/README.md b/README.md
index a37c44b..7b361a0 100644
--- a/README.md
+++ b/README.md
@@ -42,6 +42,61 @@ bypass (used by the tests), and version-based invalidation; and
 [RELEASING.md](RELEASING.md) for how the single-source version (`_version.py`) is
 bumped from the release tag.
 
+## 🕸️ CATMAID pass-through
+
+VFB hosts public, read-only CATMAID servers for several connectomics datasets
+(FAFB, FANC, L1EM, ...; registry at
+<https://virtualflybrain.org/data/EM/catmaid.json>). `vfbquery` exposes their
+query API directly, and anywhere a command takes skeleton ids you can pass
+CATMAID skids, VFB ids (`VFB_xxxxxxxx`) or a mixed list — VFB ids are converted
+through the knowledge graph's cross-references before the request is made.
+
+```python
+import vfbquery as vfb
+from vfbquery import catmaid
+
+vfb.list_catmaid_instances()          # hosted instances + tokens + projects
+vfb.list_catmaid_commands()           # the curated read-only command registry
+
+fafb = catmaid('fafb')                # optionally catmaid('fanc', project=2)
+fafb.commands()                       # {command: doc}
+
+# Mixed VFB ids and skids; the envelope carries the id mapping both ways:
+fafb.connectivity(ids=['VFB_001011rj', 10603863], boolean_op='OR')
+fafb.neuron_names(ids=['VFB_001011rj'])
+fafb.swc(id='VFB_001011rj')           # single-id commands take id=
+fafb.swc_alignments(id='VFB_001011rj')  # spaces an SWC is available in
+fafb.swc(id='VFB_001011rj', aligned='JRC2018Unisex')  # VFB's template-registered copy
+#   (aligned=<template short_form or label>; aligned='vfb' while there is only
+#    one registration; aligned='original' or omitted = the CATMAID original)
+
+# Untouched CATMAID response instead of the VFB envelope:
+fafb.neuron_names(ids=['VFB_001011rj'], raw=True)
+```
+
+By default results come back wrapped as
+`{instance, project_id, command, xref_db, id_map, unmatched, reverse_map,
+result}` where `result` is the untouched CATMAID payload, `id_map` maps the
+VFB ids you passed to skids, and `reverse_map` maps skids found in the result
+back to VFB ids where the knowledge graph knows them. `raw=True` skips the
+envelope. Parameters not interpreted by the pass-through are forwarded to
+CATMAID verbatim (use CATMAID's own parameter names, e.g. `boolean_op='AND'`).
+
+Only instances with skid cross-references in the KB (currently the FAFB, FANC
+and L1EM projects) can take VFB ids; the others work with plain skids.
+larva1099 will gain xrefs once its neurons are loaded into VFB; abd1.5,
+iav-robo, iav-tnt and l3vnc are mutant specimens, which sit outside VFB's
+wildtype-based data model, so they are expected to stay skid-only.
+
+The same surface is served over HTTP by the HA API (`python -m
+vfbquery.ha_api`):
+
+```
+GET /catmaid                                    # hosted instances
+GET /catmaid/{instance}                         # metadata + commands
+GET /catmaid/{instance}/{command}?ids=VFB_001011rj,10603863[&project=1][&raw=true]
+```
+
 To get term info for a term:
 get_term_info(ID)
 
diff --git a/src/test/test_catmaid_passthrough.py b/src/test/test_catmaid_passthrough.py
new file mode 100644
index 0000000..15e8c0f
--- /dev/null
+++ b/src/test/test_catmaid_passthrough.py
@@ -0,0 +1,504 @@
+"""Tests for the CATMAID pass-through (catmaid_client + the /catmaid routes).
+
+Unit tests exercise the registry, id handling and request assembly against
+mocks; the ``integration``-marked tests at the bottom go to the live hosted
+CATMAID instances and the KB, like the rest of this suite does. Skip those
+with ``-m 'not integration'``.
+"""
+
+import pytest
+
+import vfbquery.catmaid_client as cm
+import vfbquery.ha_api as ha_api
+
+
+# ---------------------------------------------------------------------------
+# Registry integrity
+# ---------------------------------------------------------------------------
+
+_WRITE_FRAGMENTS = ("rename", "import", "delete", "add", "fork", "revoke",
+                    "datastores", "samplers", "project-tokens", "favorite")
+
+
+def test_registry_is_read_only():
+    """No command may point at a CATMAID write or admin endpoint."""
+    for name, spec in cm.CATMAID_COMMANDS.items():
+        assert spec["method"] in ("GET", "POST"), name
+        for fragment in _WRITE_FRAGMENTS:
+            assert fragment not in spec["path"], (name, fragment)
+
+
+def test_registry_paths_format_cleanly():
+    slots = {"project_id": 1, "skeleton_id": 2, "neuron_id": 3,
+             "treenode_id": 4, "connector_id": 5, "node_type": "treenode",
+             "node_id": 6}
+    for name, spec in cm.CATMAID_COMMANDS.items():
+        if spec.get("local"):        # answered by VFBquery, not CATMAID
+            continue
+        path = spec["path"].format(**slots)
+        assert "{" not in path and "}" not in path, name
+        assert path.startswith("/"), name
+
+
+def test_registry_id_specs_are_well_formed():
+    for name, spec in cm.CATMAID_COMMANDS.items():
+        for public, wire in (spec.get("id_params") or {}).items():
+            assert "{i}" in wire, (name, public)
+        if spec.get("id_path"):
+            assert spec["id_path"]["kind"] in ("skid", "neuron_id"), name
+            if not spec.get("local"):
+                assert "{%s}" % spec["id_path"]["slot"] in spec["path"], name
+        assert spec.get("returns", "json") in ("json", "text", "bytes"), name
+
+
+def test_list_catmaid_commands_shape():
+    listing = cm.list_catmaid_commands()
+    assert listing["connectivity"]["takes_ids"] == ["ids"]
+    assert sorted(listing["connectivity_matrix"]["takes_ids"]) == [
+        "columns", "rows"]
+    assert listing["swc"]["returns"] == "text"
+    assert listing["neuron_skeletons"]["takes_ids"] == ["id"]
+
+
+# ---------------------------------------------------------------------------
+# id parsing / classification
+# ---------------------------------------------------------------------------
+
+def test_as_id_list_accepts_mixed_forms():
+    assert cm._as_id_list("16, 17") == ["16", "17"]
+    assert cm._as_id_list(["VFB_00101567", 16]) == ["VFB_00101567", "16"]
+    assert cm._as_id_list(16) == ["16"]
+    assert cm._as_id_list(None) == []
+
+
+def test_as_id_list_rejects_junk():
+    with pytest.raises(ValueError):
+        cm._as_id_list("DROP TABLE")
+    with pytest.raises(ValueError):
+        cm._as_id_list("VFB_001")          # too short to be a VFB id
+    with pytest.raises(ValueError):
+        cm._as_id_list("VFB_00101567'")    # quote never reaches Cypher
+
+
+def test_as_id_list_caps_volume(monkeypatch):
+    monkeypatch.setattr(cm, "MAX_IDS_PER_CALL", 3)
+    with pytest.raises(ValueError):
+        cm._as_id_list(["1", "2", "3", "4"])
+
+
+def test_collect_skid_like_keys_bounded():
+    found = set()
+    payload = {"incoming": {str(i): {"x": 1} for i in range(50)},
+               "not_a_skid": {"abc": 1}}
+    cm._collect_skid_like_keys(payload, found, 10)
+    assert len(found) == 10
+    assert all(k.isdigit() for k in found)
+
+
+# ---------------------------------------------------------------------------
+# Request assembly against a mocked instance
+# ---------------------------------------------------------------------------
+
+_FAKE_CONFIG = {
+    "instances": [
+        {"id": "fafb", "name": "FAFB",
+         "url": "https://fafb.example.org", "api_token": "tok",
+         "projects": [{"id": 1, "title": "Adult Brain"}]},
+        {"id": "bare", "name": "No xrefs",
+         "url": "https://bare.example.org", "api_token": "tok2",
+         "projects": [{"id": 2, "title": "Only project"}]},
+    ]
+}
+
+
+@pytest.fixture
+def fake_instance(monkeypatch):
+    """A CatmaidInstance wired to a fake config, fake KB and captured HTTP."""
+    calls = []
+
+    monkeypatch.setattr(cm, "get_catmaid_config", lambda **kw: _FAKE_CONFIG)
+    monkeypatch.setattr(cm, "_xref_site_map",
+                        lambda **kw: {("fafb", 1): "catmaid_fafb"})
+    monkeypatch.setattr(
+        cm, "vfb_ids_to_skids",
+        lambda ids, site: {i: str(100 + n) for n, i in enumerate(sorted(ids))
+                           if not i.endswith("zz")})
+    monkeypatch.setattr(
+        cm, "skids_to_vfb_ids",
+        lambda skids, site: {s: "VFB_%08d" % int(s) for s in skids})
+
+    def fake_request(self, method, path, params):
+        calls.append((method, path, dict(params)))
+        return {"body": {"101": "a neuron"}, "kind": "json"}
+
+    monkeypatch.setattr(cm.CatmaidInstance, "_request", fake_request)
+    return cm.catmaid("fafb"), calls
+
+
+def test_mixed_ids_become_indexed_wire_params(fake_instance):
+    fafb, calls = fake_instance
+    envelope = fafb.neuron_names(ids=["VFB_0010000a", "555"])
+    method, path, params = calls[-1]
+    assert (method, path) == ("POST", "/1/skeleton/neuronnames")
+    assert params == {"skids[0]": "100", "skids[1]": "555"}
+    assert envelope["id_map"] == {"VFB_0010000a": "100"}
+    assert envelope["unmatched"] == []
+    assert envelope["xref_db"] == "catmaid_fafb"
+    assert envelope["reverse_map"]  # populated via skids_to_vfb_ids
+
+
+def test_unmatched_vfb_ids_are_reported_not_sent(fake_instance):
+    fafb, calls = fake_instance
+    envelope = fafb.neuron_names(ids=["VFB_001000zz", "555"])
+    _, _, params = calls[-1]
+    assert params == {"skids[0]": "555"}
+    assert envelope["unmatched"] == ["VFB_001000zz"]
+
+
+def test_two_id_list_params(fake_instance):
+    fafb, calls = fake_instance
+    fafb.connectivity_matrix(rows=["1"], columns=["2", "3"])
+    _, path, params = calls[-1]
+    assert path == "/1/skeleton/connectivity_matrix"
+    assert params == {"rows[0]": "1", "columns[0]": "2", "columns[1]": "3"}
+
+
+def test_passthrough_params_forwarded_verbatim(fake_instance):
+    fafb, calls = fake_instance
+    fafb.connectivity(ids=["555"], boolean_op="OR", with_nodes=False)
+    _, _, params = calls[-1]
+    assert params["boolean_op"] == "OR"
+    assert params["with_nodes"] == "false"
+
+
+def test_raw_returns_untouched_body(fake_instance):
+    fafb, _ = fake_instance
+    assert fafb.neuron_names(ids=["555"], raw=True) == {"101": "a neuron"}
+
+
+def test_vfb_ids_refused_where_no_xref_site(fake_instance):
+    _, _ = fake_instance
+    bare = cm.catmaid("bare")
+    with pytest.raises(ValueError, match="no VFB skid cross-references"):
+        bare.neuron_names(ids=["VFB_0010000a"])
+    # plain skids still fine
+    assert bare.neuron_names(ids=["555"], raw=True) == {"101": "a neuron"}
+
+
+def test_unknown_instance_and_project_and_command(fake_instance):
+    _, _ = fake_instance
+    with pytest.raises(ValueError, match="Unknown CATMAID instance"):
+        cm.catmaid("nope")
+    with pytest.raises(ValueError, match="has no project"):
+        cm.catmaid("fafb", project=9)
+    fafb = cm.catmaid("fafb")
+    with pytest.raises(ValueError, match="Unknown CATMAID command"):
+        fafb.call("write_all_the_things")
+
+
+def test_missing_required_ids(fake_instance):
+    fafb, _ = fake_instance
+    with pytest.raises(ValueError, match="requires 'ids'"):
+        fafb.connectivity()
+    with pytest.raises(ValueError, match="requires 'id'"):
+        fafb.swc()
+
+
+def test_neuron_id_bridge(monkeypatch, fake_instance):
+    fafb, calls = fake_instance
+
+    def fake_request(self, method, path, params):
+        calls.append((method, path, dict(params)))
+        if path.endswith("/neurons/from-models"):
+            return {"body": {"555": 999}, "kind": "json"}
+        return {"body": [555], "kind": "json"}
+
+    monkeypatch.setattr(cm.CatmaidInstance, "_request", fake_request)
+    envelope = fafb.neuron_skeletons(id="555")
+    assert calls[-2][1] == "/1/neurons/from-models"
+    assert calls[-1][1] == "/1/neuron/999/get-all-skeletons"
+    assert any("999" in n for n in envelope["notes"])
+
+
+def test_cypher_id_quoting_only_sees_validated_ids():
+    # _quote_list is only ever fed regex-validated ids, but keep it honest.
+    assert cm._quote_list(["16", "VFB_00101567"]) == "['16', 'VFB_00101567']"
+    for ch in "'\";":
+        assert ch not in "".join(
+            c for c in cm._quote_list(["123"]) if c not in "[]', ")
+
+
+# ---------------------------------------------------------------------------
+# ha_api plumbing
+# ---------------------------------------------------------------------------
+
+def test_catmaid_paths_pass_the_allowlist():
+    assert "/catmaid" in ha_api.ALLOWED_PATHS
+    assert "/catmaid/fafb/swc".startswith(ha_api.ALLOWED_PATH_PREFIXES)
+    assert not "/catmaidx".startswith(ha_api.ALLOWED_PATH_PREFIXES)
+
+
+def test_catmaid_name_validators():
+    assert ha_api._CATMAID_INSTANCE_RE.match("abd1.5")
+    assert ha_api._CATMAID_INSTANCE_RE.match("iav-robo")
+    assert not ha_api._CATMAID_INSTANCE_RE.match("../etc")
+    assert ha_api._CATMAID_COMMAND_RE.match("neuron_names")
+    assert not ha_api._CATMAID_COMMAND_RE.match("neuron-names/")
+
+
+def test_catmaid_raw_view_unwraps_envelope_only():
+    envelope = {"command": "swc", "result": "raw text", "id_map": {}}
+    assert ha_api._catmaid_raw_view(envelope) == "raw text"
+    assert ha_api._catmaid_raw_view({"anything": 1}) == {"anything": 1}
+    assert ha_api._catmaid_raw_view("bare") == "bare"
+
+
+# ---------------------------------------------------------------------------
+# Live integration — hosted CATMAID + KB, like the rest of the suite
+# ---------------------------------------------------------------------------
+
+@pytest.mark.integration
+def test_live_instance_listing():
+    listing = cm.list_catmaid_instances()
+    ids = {i["id"] for i in listing["instances"]}
+    assert {"fafb", "l1em", "fanc"} <= ids
+    fafb = next(i for i in listing["instances"] if i["id"] == "fafb")
+    assert fafb["projects"][0]["vfb_xref_db"] == "catmaid_fafb"
+    assert fafb["api_token"]
+
+
+@pytest.mark.integration
+def test_live_site_map_covers_both_fanc_projects():
+    site_map = cm._xref_site_map()
+    assert site_map.get(("fanc", 1)) == "catmaid_fanc"
+    assert site_map.get(("fanc", 2)) == "catmaid_fanc_JRC2018VF"
+
+
+@pytest.mark.integration
+def test_live_round_trip_on_fafb():
+    """A KB xref converts to a skid, CATMAID answers, and the reverse map
+    points back at the same VFB id."""
+    rows = cm.dict_cursor(cm._get_nc().commit_list([
+        "MATCH (s:Site {short_form: 'catmaid_fafb'})"
+        "<-[r:database_cross_reference]-(i:Entity) "
+        "RETURN i.short_form AS vfb_id, r.accession[0] AS skid LIMIT 1"]))
+    vfb_id, skid = rows[0]["vfb_id"], str(rows[0]["skid"])
+
+    fafb = cm.catmaid("fafb")
+    envelope = fafb.neuron_names(ids=[vfb_id])
+    assert envelope["id_map"] == {vfb_id: skid}
+    assert envelope["unmatched"] == []
+    assert skid in envelope["result"]
+    assert envelope["reverse_map"].get(skid) == vfb_id
+
+    raw = fafb.neuron_names(ids=[vfb_id], raw=True)
+    assert raw == envelope["result"]
+
+
+@pytest.mark.integration
+def test_live_swc_is_text():
+    rows = cm.dict_cursor(cm._get_nc().commit_list([
+        "MATCH (s:Site {short_form: 'catmaid_l1em'})"
+        "<-[r:database_cross_reference]-(i:Entity) "
+        "RETURN i.short_form AS vfb_id LIMIT 1"]))
+    l1em = cm.catmaid("l1em")
+    envelope = l1em.swc(id=rows[0]["vfb_id"])
+    lines = envelope["result"].splitlines()
+    assert lines and len(lines[0].split()) == 7   # SWC columns
+
+
+@pytest.mark.integration
+def test_live_skid_only_instance_refuses_vfb_ids():
+    l3vnc = cm.catmaid("l3vnc")
+    with pytest.raises(ValueError, match="no VFB skid cross-references"):
+        l3vnc.neuron_names(ids=["VFB_00101567"])
+    projects = l3vnc.projects(raw=True)
+    assert isinstance(projects, list) and projects
+
+
+# ---------------------------------------------------------------------------
+# Aligned SWC option
+# ---------------------------------------------------------------------------
+
+def test_truthy_understands_http_flags():
+    assert cm._truthy(True) and cm._truthy("true") and cm._truthy("1")
+    assert not cm._truthy(False) and not cm._truthy("false")
+    assert not cm._truthy("") and not cm._truthy(None)
+
+
+class _FakeResp:
+    def __init__(self, status_code=200, content=b"# SWC\n1 0 0 0 0 1 -1\n"):
+        self.status_code = status_code
+        self.content = content
+
+
+def test_aligned_swc_from_vfb_store(monkeypatch, fake_instance):
+    fafb, _ = fake_instance
+    urls = []
+    monkeypatch.setattr(cm, "list_aligned_templates", lambda vfb_id: [
+        {"template": "VFB_00101567", "label": "JRC2018Unisex",
+         "folder": "http://www.virtualflybrain.org/data/VFB/i/0010/000a/"
+                   "VFB_00101567/"}])
+
+    class FakeSession:
+        def get(self, url, **kw):
+            urls.append(url)
+            return _FakeResp()
+
+    monkeypatch.setattr(cm, "_http_session", lambda: FakeSession())
+    envelope = fafb.swc(id="VFB_0010000a", aligned=True)
+    assert envelope["aligned"] is True
+    assert envelope["template"]["label"] == "JRC2018Unisex"
+    assert envelope["result"].startswith("# SWC")
+    assert urls == ["https://www.virtualflybrain.org/data/VFB/i/0010/000a/"
+                    "VFB_00101567/volume.swc"]           # https swap + file
+    assert fafb.swc(id="VFB_0010000a", aligned="true",
+                    raw=True).startswith("# SWC")
+
+
+def test_aligned_swc_multiple_templates_need_a_choice(monkeypatch,
+                                                      fake_instance):
+    fafb, _ = fake_instance
+    regs = [{"template": "VFB_1", "label": "JRC2018Unisex",
+             "folder": "http://x/a/"},
+            {"template": "VFB_2", "label": "JRC2018UnifiedCNS",
+             "folder": "http://x/b/"}]
+    monkeypatch.setattr(cm, "list_aligned_templates", lambda vfb_id: regs)
+    # generic "the VFB copy" is ambiguous once there are two registrations
+    with pytest.raises(ValueError, match="pass aligned="):
+        fafb.swc(id="VFB_0010000a", aligned=True)
+    with pytest.raises(ValueError, match="pass aligned="):
+        fafb.swc(id="VFB_0010000a", aligned="vfb")
+
+    class FakeSession:
+        def get(self, url, **kw):
+            assert url == "https://x/b/volume.swc"
+            return _FakeResp()
+
+    monkeypatch.setattr(cm, "_http_session", lambda: FakeSession())
+    # pick by template short_form or by label, case-insensitively
+    envelope = fafb.swc(id="VFB_0010000a", aligned="VFB_2")
+    assert envelope["template"]["short_form"] == "VFB_2"
+    envelope = fafb.swc(id="VFB_0010000a", aligned="jrc2018unifiedcns")
+    assert envelope["template"]["short_form"] == "VFB_2"
+    with pytest.raises(ValueError, match="not registered to template"):
+        fafb.swc(id="VFB_0010000a", aligned="VFB_9")
+
+
+def test_aligned_swc_rejects_stray_params_and_id_lists(fake_instance):
+    fafb, _ = fake_instance
+    with pytest.raises(ValueError, match="takes only id="):
+        fafb.swc(id="VFB_0010000a", aligned=True, with_tags=1)
+    with pytest.raises(ValueError, match="exactly one id"):
+        fafb.swc(id="VFB_0010000a,VFB_0010000b", aligned=True)
+
+
+def test_aligned_original_means_catmaid(fake_instance):
+    fafb, calls = fake_instance
+    fafb.swc(id="555", aligned="original", raw=True)
+    method, path, _ = calls[-1]
+    assert (method, path) == ("GET", "/1/skeleton/555/swc")
+    fafb.swc(id="555", aligned="false", raw=True)
+    assert calls[-1][1] == "/1/skeleton/555/swc"
+
+
+def test_catmaid_config_cached_for_whole_run_by_default(monkeypatch):
+    fetches = []
+
+    class FakeSession:
+        def get(self, url, **kw):
+            fetches.append(url)
+            resp = _FakeResp()
+            resp.raise_for_status = lambda: None
+            resp.json = lambda: {"instances": [{"id": "x"}]}
+            return resp
+
+    monkeypatch.setattr(cm, "_http_session", lambda: FakeSession())
+    monkeypatch.setattr(cm, "CATMAID_CONFIG_TTL", 0.0)
+    monkeypatch.setattr(cm, "_config_cache", {"fetched": 0.0, "data": None})
+    assert cm.get_catmaid_config()["instances"] == [{"id": "x"}]
+    assert cm.get_catmaid_config()["instances"] == [{"id": "x"}]
+    assert len(fetches) == 1                     # one fetch per process
+    cm.get_catmaid_config(force_refresh=True)
+    assert len(fetches) == 2
+
+
+@pytest.mark.integration
+def test_live_aligned_swc_on_fafb():
+    fafb = cm.catmaid("fafb")
+    envelope = fafb.swc(id="VFB_0010009u", aligned=True)
+    assert envelope["aligned"] is True
+    assert envelope["template"]["short_form"] == "VFB_00101567"  # JRC2018U
+    assert envelope["result"].lstrip().startswith("#")
+    original = fafb.swc(id="VFB_0010009u", raw=True)
+    assert original[:200] != envelope["result"][:200]   # different space
+
+
+# ---------------------------------------------------------------------------
+# swc_alignments
+# ---------------------------------------------------------------------------
+
+def test_swc_alignments_lists_original_plus_vfb(monkeypatch, fake_instance):
+    fafb, _ = fake_instance
+    monkeypatch.setattr(cm, "list_aligned_templates", lambda vfb_id: [
+        {"template": "VFB_1", "label": "JRC2018Unisex",
+         "folder": "http://x/a/"}])
+
+    class FakeSession:
+        def head(self, url, **kw):
+            assert url == "https://x/a/volume.swc"
+            return _FakeResp()
+
+    monkeypatch.setattr(cm, "_http_session", lambda: FakeSession())
+    envelope = fafb.swc_alignments(id="VFB_0010000a")
+    rows = envelope["result"]["alignments"]
+    assert rows[0]["aligned"] == "original"
+    assert rows[0]["source"] == "catmaid" and rows[0]["swc_available"]
+    assert rows[1]["aligned"] == "VFB_1" and rows[1]["source"] == "vfb"
+    assert rows[1]["swc_available"] is True
+    assert envelope["result"]["skid"] == "100"      # from the fake xref map
+    assert envelope["id_map"] == {"VFB_0010000a": "100"}
+
+
+def test_swc_alignments_skid_without_vfb_record(monkeypatch, fake_instance):
+    fafb, _ = fake_instance
+    monkeypatch.setattr(cm, "skids_to_vfb_ids", lambda skids, site: {})
+    envelope = fafb.swc_alignments(id="555")
+    rows = envelope["result"]["alignments"]
+    assert [r["aligned"] for r in rows] == ["original"]
+    assert envelope["result"]["vfb_id"] is None
+    assert any("no VFB record" in n for n in envelope["notes"])
+
+
+def test_swc_alignments_marks_missing_store_files(monkeypatch, fake_instance):
+    fafb, _ = fake_instance
+    monkeypatch.setattr(cm, "list_aligned_templates", lambda vfb_id: [
+        {"template": "VFB_1", "label": "A", "folder": "http://x/a/"}])
+
+    class FakeSession:
+        def head(self, url, **kw):
+            return _FakeResp(status_code=404)
+
+    monkeypatch.setattr(cm, "_http_session", lambda: FakeSession())
+    rows = fafb.swc_alignments(id="VFB_0010000a", raw=True)["alignments"]
+    assert rows[1]["swc_available"] is False
+
+
+def test_swc_alignments_input_validation(fake_instance):
+    fafb, _ = fake_instance
+    with pytest.raises(ValueError, match="takes only id="):
+        fafb.swc_alignments(id="555", nonsense=1)
+    with pytest.raises(ValueError, match="exactly one id"):
+        fafb.swc_alignments(id="1,2")
+
+
+@pytest.mark.integration
+def test_live_swc_alignments_on_fafb():
+    fafb = cm.catmaid("fafb")
+    envelope = fafb.swc_alignments(id="VFB_0010009u")
+    rows = envelope["result"]["alignments"]
+    assert rows[0]["aligned"] == "original"
+    vfb_rows = [r for r in rows if r["source"] == "vfb"]
+    assert any(r["template"] == "VFB_00101567" and r["swc_available"]
+               for r in vfb_rows)                     # JRC2018U copy exists
+    assert envelope["result"]["skid"] == "13146"
diff --git a/src/vfbquery/__init__.py b/src/vfbquery/__init__.py
index ffbee01..ed77cd9 100644
--- a/src/vfbquery/__init__.py
+++ b/src/vfbquery/__init__.py
@@ -4,6 +4,8 @@
 from .flybase_combo_pubs import resolve_combination, find_combo_publications
 from .vfb_connectivity import list_connectome_datasets, query_connectivity
 from .graph_builder import build_graph, batch_lookup_ids
+from .catmaid_client import (catmaid, CatmaidInstance, list_catmaid_instances,
+                             list_catmaid_commands, run_catmaid_command)
 
 # SOLR-based caching (simplified single-layer approach)
 try:
diff --git a/src/vfbquery/catmaid_client.py b/src/vfbquery/catmaid_client.py
new file mode 100644
index 0000000..c129def
--- /dev/null
+++ b/src/vfbquery/catmaid_client.py
@@ -0,0 +1,1083 @@
+"""
+CATMAID pass-through for the VFB-hosted CATMAID instances.
+
+VFB hosts public, read-only CATMAID servers for several connectomics
+datasets (FAFB, FANC, L1EM, ...). Their connection details — base URL,
+anonymous API token and project list — are published at
+https://virtualflybrain.org/data/EM/catmaid.json and are fetched (and
+cached) from there at runtime, so new instances appear here without a
+code change. The tokens are public by design: they authenticate as
+CATMAID's AnonymousUser, whose only permission is ``can_browse``, so
+they grant exactly the read access the servers already offer everyone
+and the server refuses writes made with them.
+
+What this module adds over calling CATMAID directly:
+
+* A curated, read-only command registry (:data:`CATMAID_COMMANDS`)
+  covering the sensible query surface of the CATMAID HTTP API —
+  skeletons, connectivity, annotations, connectors, labels, nodes,
+  stats — with writes and admin endpoints deliberately absent.
+* VFB id handling: anywhere a command takes skeleton ids, callers may
+  pass CATMAID skeleton ids (skids), VFB short_form ids (``VFB_xxxxxxxx``)
+  or a mixed list. VFB ids are converted to skids through the knowledge
+  graph's ``database_cross_reference`` xrefs before the request is made,
+  and the response envelope carries the mapping both ways.
+* CATMAID neuron ids are not stored in the VFB KB, so commands that
+  address a *neuron* (rather than a skeleton) transparently derive the
+  neuron id from the skid via CATMAID's ``neurons/from-models``.
+
+By default results come back in a VFB envelope::
+
+    {
+      "instance":  "fafb",
+      "project_id": 1,
+      "command":   "connectivity",
+      "xref_db":   "catmaid_fafb",        # KB site used for id conversion
+      "id_map":    {"VFB_001011rj": "2856545"},  # input VFB id -> skid
+      "unmatched": [],                      # inputs that could not be mapped
+      "reverse_map": {"2856545": "VFB_001011rj"},# skids seen in result -> VFB id
+      "result":    <untouched CATMAID response>
+    }
+
+Pass ``raw=True`` (or ``?raw=true`` over HTTP) to get the untouched
+CATMAID response alone.
+
+Only some instances have skid xrefs in the VFB KB (currently the FAFB,
+FANC and L1EM projects). On instances without xrefs the pass-through
+still works with plain skids; passing a VFB id there raises a clear
+error instead of guessing. larva1099 will gain xrefs once its neurons
+are loaded into VFB; the remaining instances (abd1.5, iav-robo, iav-tnt,
+l3vnc) host mutant specimens, which sit outside VFB's wildtype-based
+data model, so they are expected to stay skid-only.
+"""
+
+import json
+import base64
+import logging
+import os
+import re
+import threading
+import time
+from urllib.parse import urlparse
+
+import requests
+
+from .neo4j_client import Neo4jConnect, dict_cursor
+
+log = logging.getLogger("vfbquery.catmaid")
+
+# ---------------------------------------------------------------------------
+# Configuration / constants
+# ---------------------------------------------------------------------------
+
+#: Where the public instance registry lives. Override for testing.
+CATMAID_JSON_URL = os.getenv(
+    "VFBQUERY_CATMAID_JSON_URL",
+    "https://virtualflybrain.org/data/EM/catmaid.json")
+
+#: How long the fetched catmaid.json is trusted, seconds. 0 (the default)
+#: means for the whole run: instances are added rarely and tokens only ever
+#: fail loudly (401), so one fetch per process is enough. Set a positive
+#: TTL for a long-lived server that should pick up new instances without a
+#: restart, or use get_catmaid_config(force_refresh=True).
+CATMAID_CONFIG_TTL = float(os.getenv("VFBQUERY_CATMAID_CONFIG_TTL", "0"))
+
+#: How long the KB-derived (instance, project) -> xref-site map is trusted.
+CATMAID_SITES_TTL = float(os.getenv("VFBQUERY_CATMAID_SITES_TTL", "3600"))
+
+#: HTTP timeouts for calls to the CATMAID servers (connect, read).
+CATMAID_TIMEOUT = (10.0, float(os.getenv("VFBQUERY_CATMAID_READ_TIMEOUT_S", "120")))
+
+#: Hard cap on how many ids one call may convert / pass through.
+MAX_IDS_PER_CALL = int(os.getenv("VFBQUERY_CATMAID_MAX_IDS", "2000"))
+
+#: Cap on how many distinct skid-shaped result keys the reverse lookup
+#: will try to map back to VFB ids.
+MAX_REVERSE_LOOKUP = int(os.getenv("VFBQUERY_CATMAID_MAX_REVERSE", "2000"))
+
+_VFB_ID_RE = re.compile(r"^VFB_[A-Za-z0-9]{8}$")
+_SKID_RE = re.compile(r"^\d+$")
+
+#: Fallback (instance_id, project_id) -> KB Site short_form map, used only
+#: if the live KB lookup fails. Derived from the KB on 2026-08-29.
+_STATIC_XREF_SITES = {
+    ("fafb", 1): "catmaid_fafb",
+    ("fanc", 1): "catmaid_fanc",
+    ("fanc", 2): "catmaid_fanc_JRC2018VF",
+    ("l1em", 1): "catmaid_l1em",
+}
+
+
+# ---------------------------------------------------------------------------
+# Command registry — the curated, read-only CATMAID query surface.
+#
+# Each entry:
+#   method   — HTTP verb used against CATMAID.
+#   path     — path template; {project_id} always available, plus optional
+#              {skeleton_id} / {neuron_id} / other slots.
+#   doc      — one-line description (surfaced by list_catmaid_commands()).
+#   id_params— mapping of PUBLIC parameter name -> wire template for a
+#              *list* of skeleton ids ("skids[{i}]" style indexed arrays).
+#              Values accept skids, VFB ids, or a mixed list.
+#   id_path  — a single id in the path: {"slot": <path slot>,
+#              "kind": "skid" | "neuron_id"}. Public name is always "id".
+#              kind "neuron_id" accepts a skid/VFB id and bridges to the
+#              CATMAID neuron id via neurons/from-models.
+#   returns  — "json" (default), "text" or "bytes".
+#
+# Anything else the caller passes is forwarded to CATMAID verbatim (use
+# CATMAID's own parameter names, including indexed forms like
+# "annotated_with[0]" where the API wants arrays), so options that are not
+# listed here still work. Write/admin endpoints are deliberately absent —
+# the anonymous tokens cannot use them anyway.
+# ---------------------------------------------------------------------------
+
+CATMAID_COMMANDS = {
+    # -- instance-level -----------------------------------------------------
+    "projects": {
+        "method": "GET", "path": "/projects/",
+        "doc": "List projects visible on this instance.", "project": False},
+    "annotations": {
+        "method": "GET", "path": "/{project_id}/annotations/",
+        "doc": "List annotations in the project."},
+    "labels": {
+        "method": "GET", "path": "/{project_id}/labels/",
+        "doc": "List all (treenode) labels in use."},
+    "label_stats": {
+        "method": "GET", "path": "/{project_id}/labels/stats",
+        "doc": "Label usage statistics."},
+    "stats_nodecount": {
+        "method": "GET", "path": "/{project_id}/stats/nodecount",
+        "doc": "Nodes created per user."},
+    "stats_cable_length": {
+        "method": "GET", "path": "/{project_id}/stats/cable-length",
+        "doc": "Largest skeletons by cable length."},
+    "stats_server": {
+        "method": "GET", "path": "/{project_id}/stats/server",
+        "doc": "Server state information."},
+    "origins": {
+        "method": "GET", "path": "/{project_id}/origins/",
+        "doc": "List available data sources."},
+    "interpolatable_sections": {
+        "method": "GET", "path": "/{project_id}/interpolatable-sections/",
+        "doc": "Broken/interpolatable section locations."},
+    "deep_links": {
+        "method": "GET", "path": "/{project_id}/links/",
+        "doc": "List saved deep links."},
+    "landmarks": {
+        "method": "GET", "path": "/{project_id}/landmarks/",
+        "doc": "List landmarks (with_locations=true for coordinates)."},
+    "landmark_groups": {
+        "method": "GET", "path": "/{project_id}/landmarks/groups/",
+        "doc": "List landmark groups (with_members/with_locations options)."},
+    "similarity_configs": {
+        "method": "GET", "path": "/{project_id}/similarity/configs/",
+        "doc": "List NBLAST similarity configurations."},
+    "similarity_queries": {
+        "method": "GET", "path": "/{project_id}/similarity/queries/",
+        "doc": "List NBLAST similarity tasks."},
+    "connector_types": {
+        "method": "GET", "path": "/{project_id}/connectors/types/",
+        "doc": "List available connector (synapse) link types."},
+
+    # -- finding things -----------------------------------------------------
+    "list_skeletons": {
+        "method": "GET", "path": "/{project_id}/skeletons/",
+        "doc": "List skeleton ids by filter (nodecount_gt=, created_by=, ...)."},
+    "list_neurons": {
+        "method": "GET", "path": "/{project_id}/neurons/",
+        "doc": "List neurons by filter criteria."},
+    "annotations_query_targets": {
+        "method": "POST", "path": "/{project_id}/annotations/query-targets",
+        "doc": "Find neurons/annotations by annotation or name "
+               "(name=, annotated_with=, types[0]=neuron, ...)."},
+    "find_label_nodes": {
+        "method": "POST", "path": "/{project_id}/nodes/find-labels",
+        "doc": "Find nodes whose labels match a query (query=)."},
+    "nearest_node": {
+        "method": "GET", "path": "/{project_id}/nodes/nearest",
+        "doc": "Closest node to a location (x=, y=, z=)."},
+
+    # -- skeleton queries (accept skids and/or VFB ids) ---------------------
+    "neuron_names": {
+        "method": "POST", "path": "/{project_id}/skeleton/neuronnames",
+        "doc": "Map skeleton ids to neuron names.",
+        "id_params": {"ids": "skids[{i}]"}},
+    "skeleton_validity": {
+        "method": "POST", "path": "/{project_id}/skeletons/validity",
+        "doc": "Which of the given skeleton ids exist in the project.",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "skeleton_summary": {
+        "method": "POST", "path": "/{project_id}/skeletons/summary",
+        "doc": "Summary information (node counts, cable, review) per skeleton.",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "cable_length": {
+        "method": "POST", "path": "/{project_id}/skeletons/cable-length",
+        "doc": "Cable length per skeleton.",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "review_status": {
+        "method": "POST", "path": "/{project_id}/skeletons/review-status",
+        "doc": "Review status per skeleton.",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "compact_detail": {
+        "method": "POST", "path": "/{project_id}/skeletons/compact-detail",
+        "doc": "Compact treenode representation for a set of skeletons "
+               "(with_connectors=, with_tags=, ...).",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "annotations_for_skeletons": {
+        "method": "POST", "path": "/{project_id}/annotations/forskeletons",
+        "doc": "Annotations on each of a set of skeletons.",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "connectivity": {
+        "method": "POST", "path": "/{project_id}/skeletons/connectivity",
+        "doc": "Upstream/downstream synaptic partners "
+               "(boolean_op=OR, with_nodes=false, ...).",
+        "id_params": {"ids": "source_skeleton_ids[{i}]"}},
+    "connectivity_counts": {
+        "method": "POST", "path": "/{project_id}/skeletons/connectivity-counts",
+        "doc": "Synapse counts by link type per skeleton.",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "connectivity_matrix": {
+        "method": "POST", "path": "/{project_id}/skeleton/connectivity_matrix",
+        "doc": "Sparse connectivity matrix between two skeleton sets.",
+        "id_params": {"rows": "rows[{i}]", "columns": "columns[{i}]"}},
+    "circles_of_hell": {
+        "method": "POST", "path": "/{project_id}/graph/circlesofhell",
+        "doc": "Skeletons within n hops of the given set (n_circles=1, ...).",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "connector_links": {
+        "method": "POST", "path": "/{project_id}/connectors/links/",
+        "doc": "Connector links on a set of skeletons "
+               "(relation_type=presynaptic_to|postsynaptic_to|...).",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+    "neuron_ids": {
+        "method": "POST", "path": "/{project_id}/neurons/from-models",
+        "doc": "CATMAID neuron id for each skeleton id.",
+        "id_params": {"ids": "model_ids[{i}]"}},
+    "sampler_count": {
+        "method": "POST", "path": "/{project_id}/skeletons/sampler-count",
+        "doc": "Number of reconstruction samplers per skeleton.",
+        "id_params": {"ids": "skeleton_ids[{i}]"}},
+
+    # -- single-skeleton queries (id= accepts one skid or VFB id) -----------
+    "swc": {
+        "method": "GET", "path": "/{project_id}/skeleton/{skeleton_id}/swc",
+        "doc": "Skeleton as SWC text. aligned=<template short_form or label> "
+               "returns VFB's copy registered to that template instead of "
+               "the original EM-space skeleton (aligned=vfb picks the VFB "
+               "copy while there is only one; aligned=original or omitted "
+               "is the CATMAID original).",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}, "returns": "text"},
+    "swc_alignments": {
+        "method": "GET", "path": "(served by VFBquery, not CATMAID)",
+        "local": True,
+        "doc": "List the spaces an swc download is available in for one "
+               "neuron: the original EM space plus any VFB template-"
+               "registered copies. Each row carries the aligned= value to "
+               "pass to swc.",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}},
+    "eswc": {
+        "method": "GET", "path": "/{project_id}/skeleton/{skeleton_id}/eswc",
+        "doc": "Skeleton as extended SWC text (creator/edit metadata).",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}, "returns": "text"},
+    "neuroglancer_skeleton": {
+        "method": "GET",
+        "path": "/{project_id}/skeletons/{skeleton_id}/neuroglancer",
+        "doc": "Skeleton in neuroglancer precomputed format "
+               "(base64 in the JSON envelope).",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}, "returns": "bytes"},
+    "skeleton_root": {
+        "method": "GET", "path": "/{project_id}/skeletons/{skeleton_id}/root",
+        "doc": "Root treenode id and location of a skeleton.",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}},
+    "skeleton_cable_length": {
+        "method": "GET",
+        "path": "/{project_id}/skeletons/{skeleton_id}/cable-length",
+        "doc": "Cable length of a single skeleton.",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}},
+    "skeleton_compact_detail": {
+        "method": "GET",
+        "path": "/{project_id}/skeletons/{skeleton_id}/compact-detail",
+        "doc": "Compact treenode representation of one skeleton "
+               "(with_connectors=, with_tags=, ...).",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}},
+    "skeleton_node_overview": {
+        "method": "GET",
+        "path": "/{project_id}/skeletons/{skeleton_id}/node-overview",
+        "doc": "Treenode / review / label overview of one skeleton.",
+        "id_path": {"slot": "skeleton_id", "kind": "skid"}},
+    "neuron_skeletons": {
+        "method": "GET",
+        "path": "/{project_id}/neuron/{neuron_id}/get-all-skeletons",
+        "doc": "All skeleton ids modelling a neuron. id= accepts a skid or "
+               "VFB id; the CATMAID neuron id is derived automatically.",
+        "id_path": {"slot": "neuron_id", "kind": "neuron_id"}},
+
+    # -- nodes / treenodes / connectors -------------------------------------
+    "node_locations": {
+        "method": "POST", "path": "/{project_id}/nodes/location",
+        "doc": "Locations for a set of node ids (node_ids[0]=..., raw node "
+               "ids, not skeleton ids)."},
+    "treenode_info": {
+        "method": "GET", "path": "/{project_id}/treenodes/{treenode_id}/info",
+        "doc": "Skeleton/neuron information for one treenode "
+               "(path id is a raw treenode id).",
+        "path_params": ["treenode_id"]},
+    "connector_info": {
+        "method": "GET", "path": "/{project_id}/connectors/{connector_id}/",
+        "doc": "Detailed information on one connector "
+               "(path id is a raw connector id).",
+        "path_params": ["connector_id"]},
+    "node_labels": {
+        "method": "GET", "path": "/{project_id}/labels/{node_type}/{node_id}/",
+        "doc": "Labels on one node (node_type=treenode|connector, node_id=).",
+        "path_params": ["node_type", "node_id"]},
+
+    # -- spatial ------------------------------------------------------------
+    "skeletons_in_bounding_box": {
+        "method": "POST", "path": "/{project_id}/skeletons/in-bounding-box",
+        "doc": "Skeleton ids intersecting a bounding box "
+               "(minx=..maxz=, min_nodes=, ...)."},
+    "connectors_in_bounding_box": {
+        "method": "POST", "path": "/{project_id}/connectors/in-bounding-box",
+        "doc": "Connectors in a bounding box (minx=..maxz=, ...)."},
+    "skeletons_within_distance": {
+        "method": "POST",
+        "path": "/{project_id}/skeletons/within-spatial-distance",
+        "doc": "Skeletons with nodes within a distance of a location."},
+}
+
+
+# ---------------------------------------------------------------------------
+# catmaid.json config cache
+# ---------------------------------------------------------------------------
+
+_config_lock = threading.Lock()
+_config_cache = {"fetched": 0.0, "data": None}
+
+
+def _http_session():
+    """One requests session per process (connection pooling)."""
+    global _SESSION
+    try:
+        return _SESSION
+    except NameError:
+        _SESSION = requests.Session()
+        return _SESSION
+
+
+def get_catmaid_config(force_refresh=False):
+    """The parsed catmaid.json registry, cached for :data:`CATMAID_CONFIG_TTL`.
+
+    Fails soft: if a refresh fails but a previously fetched copy exists, the
+    stale copy is returned (a stale token fails loudly at the CATMAID end;
+    an unnecessary hard failure here would take every instance down at once).
+    """
+    now = time.monotonic()
+    with _config_lock:
+        if (not force_refresh and _config_cache["data"] is not None
+                and (CATMAID_CONFIG_TTL <= 0
+                     or now - _config_cache["fetched"] < CATMAID_CONFIG_TTL)):
+            return _config_cache["data"]
+        try:
+            resp = _http_session().get(CATMAID_JSON_URL, timeout=CATMAID_TIMEOUT)
+            resp.raise_for_status()
+            data = resp.json()
+            if not isinstance(data.get("instances"), list):
+                raise ValueError("catmaid.json has no 'instances' list")
+            _config_cache.update(fetched=now, data=data)
+            return data
+        except Exception as exc:
+            if _config_cache["data"] is not None:
+                log.warning("catmaid.json refresh failed (%s) — using stale copy",
+                            exc)
+                return _config_cache["data"]
+            raise
+
+
+def _instances_by_id(config=None):
+    config = config or get_catmaid_config()
+    return {inst["id"]: inst for inst in config.get("instances", [])
+            if inst.get("id")}
+
+
+# ---------------------------------------------------------------------------
+# KB xref-site discovery — which Site node holds skids for which
+# (instance, project), derived from the Site link_base URLs.
+# ---------------------------------------------------------------------------
+
+_NC = None
+_nc_lock = threading.Lock()
+
+
+def _get_nc():
+    """Per-process Neo4jConnect, same rationale as vfb_connectivity._get_nc."""
+    global _NC
+    with _nc_lock:
+        if _NC is None:
+            _NC = Neo4jConnect()
+        return _NC
+
+
+_sites_lock = threading.Lock()
+_sites_cache = {"fetched": 0.0, "map": None}
+
+_PID_RE = re.compile(r"[?&]pid=(\d+)")
+
+
+def _xref_site_map(force_refresh=False):
+    """{(instance_id, project_id): site_short_form} from the KB.
+
+    Site nodes for the hosted instances carry link_base URLs like
+    ``https://fafb.catmaid.virtualflybrain.org/?pid=1&...`` — the host names
+    the instance and ``pid`` the project, so the map derives itself and new
+    xref sites are picked up without a code change. Falls back to
+    :data:`_STATIC_XREF_SITES` if the KB is unreachable.
+    """
+    now = time.monotonic()
+    with _sites_lock:
+        if (not force_refresh and _sites_cache["map"] is not None
+                and now - _sites_cache["fetched"] < CATMAID_SITES_TTL):
+            return _sites_cache["map"]
+        query = (
+            "MATCH (s:Site) WHERE NOT s:Deprecated "
+            "AND ANY(lb IN s.link_base WHERE lb CONTAINS "
+            "'catmaid.virtualflybrain.org') "
+            "RETURN s.short_form AS site, s.link_base[0] AS link")
+        try:
+            rows = dict_cursor(_get_nc().commit_list([query]))
+            site_map = {}
+            for row in rows or []:
+                link = row.get("link") or ""
+                host = urlparse(link).netloc.lower()
+                instance = host.split(".catmaid.")[0]
+                pid_match = _PID_RE.search(link)
+                if not instance or not pid_match:
+                    continue
+                site_map[(instance, int(pid_match.group(1)))] = row["site"]
+            if not site_map:
+                raise ValueError("no CATMAID Site nodes found")
+            _sites_cache.update(fetched=now, map=site_map)
+            return site_map
+        except Exception as exc:
+            log.warning("CATMAID xref-site lookup failed (%s) — using static "
+                        "fallback map", exc)
+            return dict(_STATIC_XREF_SITES)
+
+
+# ---------------------------------------------------------------------------
+# id classification and conversion
+# ---------------------------------------------------------------------------
+
+def _as_id_list(value):
+    """Normalise an ids argument (scalar, list, or comma-separated string)."""
+    if value is None:
+        return []
+    if isinstance(value, (list, tuple, set)):
+        items = list(value)
+    elif isinstance(value, str):
+        items = [p for p in (s.strip() for s in value.split(",")) if p]
+    else:
+        items = [value]
+    out = []
+    for item in items:
+        s = str(item).strip()
+        if not s:
+            continue
+        if not (_VFB_ID_RE.match(s) or _SKID_RE.match(s)):
+            raise ValueError(
+                "'%s' is neither a CATMAID skeleton id nor a VFB id "
+                "(VFB_xxxxxxxx)" % s)
+        out.append(s)
+    if len(out) > MAX_IDS_PER_CALL:
+        raise ValueError("Too many ids in one call (%d > %d)"
+                         % (len(out), MAX_IDS_PER_CALL))
+    return out
+
+
+def _quote_list(values):
+    """Quote validated ids as a Cypher string list literal."""
+    return "[" + ", ".join("'%s'" % v for v in values) + "]"
+
+
+def vfb_ids_to_skids(vfb_ids, site):
+    """{vfb_id: skid} for the given VFB ids on one KB xref site."""
+    vfb_ids = sorted(set(vfb_ids))
+    if not vfb_ids:
+        return {}
+    query = (
+        "MATCH (s:Site {short_form: '%s'})<-[r:database_cross_reference]"
+        "-(i:Entity) WHERE i.short_form IN %s "
+        "RETURN i.short_form AS vfb_id, r.accession[0] AS acc"
+        % (site, _quote_list(vfb_ids)))
+    rows = dict_cursor(_get_nc().commit_list([query]))
+    return {row["vfb_id"]: str(row["acc"]) for row in rows or []
+            if row.get("acc") is not None}
+
+
+def skids_to_vfb_ids(skids, site):
+    """{skid: vfb_id} for the given skids on one KB xref site."""
+    skids = sorted({str(s) for s in skids})
+    if not skids:
+        return {}
+    query = (
+        "MATCH (s:Site {short_form: '%s'})<-[r:database_cross_reference]"
+        "-(i:Entity) WHERE r.accession[0] IN %s "
+        "RETURN r.accession[0] AS acc, i.short_form AS vfb_id"
+        % (site, _quote_list(skids)))
+    rows = dict_cursor(_get_nc().commit_list([query]))
+    return {str(row["acc"]): row["vfb_id"] for row in rows or []}
+
+
+def _truthy(value):
+    """Truthiness that also understands HTTP-style string flags."""
+    if isinstance(value, str):
+        return value.strip().lower() in ("1", "true", "yes")
+    return bool(value)
+
+
+def list_aligned_templates(vfb_id):
+    """Template registrations of one VFB individual's image.
+
+    :return: list of ``{"template": short_form, "label": label,
+        "folder": url}`` — one entry per template space VFB has this
+        image registered to. The aligned SWC, where one exists, is
+        ``folder + 'volume.swc'``.
+    """
+    if not _VFB_ID_RE.match(vfb_id or ""):
+        raise ValueError("'%s' is not a VFB id" % vfb_id)
+    query = (
+        "MATCH (n:Individual {short_form: '%s'})<-[:depicts]-(c:Individual)"
+        "-[ir:in_register_with]->(tc:Individual)-[:depicts]->(t:Individual) "
+        "RETURN t.short_form AS template, t.label AS label, "
+        "ir.folder[0] AS folder" % vfb_id)
+    rows = dict_cursor(_get_nc().commit_list([query]))
+    return [{"template": r["template"], "label": r.get("label"),
+             "folder": r["folder"]} for r in rows or [] if r.get("folder")]
+
+
+def _collect_skid_like_keys(obj, found, limit):
+    """Recursively collect dict keys that look like skids (bounded)."""
+    if len(found) >= limit:
+        return
+    if isinstance(obj, dict):
+        for key, value in obj.items():
+            if isinstance(key, (str, int)) and _SKID_RE.match(str(key)):
+                found.add(str(key))
+                if len(found) >= limit:
+                    return
+            _collect_skid_like_keys(value, found, limit)
+    elif isinstance(obj, list):
+        for item in obj[:limit]:
+            _collect_skid_like_keys(item, found, limit)
+
+
+# ---------------------------------------------------------------------------
+# The instance client
+# ---------------------------------------------------------------------------
+
+class CatmaidInstance:
+    """A client for one VFB-hosted CATMAID instance.
+
+    Usually obtained via :func:`catmaid`::
+
+        import vfbquery as vfb
+        fafb = vfb.catmaid('fafb')
+        fafb.connectivity(ids=['VFB_001011rj', 10603863])
+        fafb.swc(id='VFB_001011rj', raw=True)
+
+    Every registry command is callable as a method (or through
+    :meth:`call`); extra keyword arguments are forwarded to CATMAID
+    verbatim under CATMAID's own parameter names.
+    """
+
+    def __init__(self, instance, project=None):
+        instances = _instances_by_id()
+        if instance not in instances:
+            raise ValueError(
+                "Unknown CATMAID instance '%s'. Hosted instances: %s"
+                % (instance, ", ".join(sorted(instances))))
+        self._meta = instances[instance]
+        self.instance = instance
+        self.base_url = self._meta["url"].rstrip("/")
+        self.token = self._meta.get("api_token")
+        projects = {int(p["id"]): p for p in self._meta.get("projects", [])}
+        if project is None:
+            self.project_id = min(projects) if projects else 1
+        else:
+            project = int(project)
+            if projects and project not in projects:
+                raise ValueError(
+                    "Instance '%s' has no project %d. Projects: %s"
+                    % (instance, project,
+                       ", ".join("%d (%s)" % (i, p.get("title", ""))
+                                 for i, p in sorted(projects.items()))))
+            self.project_id = project
+        #: {project_id: project dict} — deliberately NOT named ``projects``,
+        #: which is the registry command listing the instance's projects.
+        self.project_map = projects
+
+    # -- metadata -----------------------------------------------------------
+
+    @property
+    def xref_db(self):
+        """KB Site short_form holding skid xrefs for this project, or None."""
+        return _xref_site_map().get((self.instance, self.project_id))
+
+    def commands(self):
+        """{command: doc} for every command this pass-through offers."""
+        return {name: spec["doc"] for name, spec in
+                sorted(CATMAID_COMMANDS.items())}
+
+    # -- id conversion ------------------------------------------------------
+
+    def resolve_ids(self, ids):
+        """Split/convert a mixed id list.
+
+        :return: (skids, id_map, unmatched) — ``skids`` in input order where
+            possible, ``id_map`` {vfb_id: skid} for converted inputs,
+            ``unmatched`` the VFB ids that had no skid xref here.
+        """
+        items = _as_id_list(ids)
+        vfb_ids = [i for i in items if _VFB_ID_RE.match(i)]
+        id_map, unmatched = {}, []
+        if vfb_ids:
+            site = self.xref_db
+            if site is None:
+                raise ValueError(
+                    "Instance '%s' (project %d) has no VFB skid cross-"
+                    "references in the knowledge graph — pass CATMAID "
+                    "skeleton ids instead of VFB ids (%s)"
+                    % (self.instance, self.project_id, ", ".join(vfb_ids)))
+            id_map = vfb_ids_to_skids(vfb_ids, site)
+            unmatched = [v for v in vfb_ids if v not in id_map]
+        skids = [id_map.get(i, i) for i in items if i not in unmatched]
+        return skids, id_map, unmatched
+
+    def _skid_to_neuron_id(self, skid):
+        """CATMAID neuron id for one skid, via neurons/from-models."""
+        data = self._request(
+            "POST", "/%d/neurons/from-models" % self.project_id,
+            {"model_ids[0]": skid})
+        payload = data["body"]
+        if isinstance(payload, dict) and str(skid) in payload:
+            return payload[str(skid)]
+        raise ValueError("CATMAID has no neuron for skeleton id %s on '%s'"
+                         % (skid, self.instance))
+
+    # -- aligned SWC from the VFB image store -------------------------------
+
+    def _aligned_swc(self, id, choice, raw=False, extra=None):
+        """VFB's template-registered SWC for one neuron.
+
+        CATMAID serves skeletons in the dataset's own EM space; VFB also
+        stores a copy registered to a standard template (``volume.swc`` in
+        the image's ``in_register_with`` folder). ``choice`` names the
+        template — short_form (``VFB_00101567``) or label
+        (``JRC2018Unisex``), case-insensitively — or is one of the generic
+        values (``vfb``/``true``), which work only while the image is
+        registered to a single template; once a neuron has several
+        registrations (e.g. JRC2018U plus a unified brain+VNC space) the
+        generic form errors and lists the choices instead of guessing.
+        """
+        if extra:
+            raise ValueError(
+                "aligned= takes only id= — unexpected: %s"
+                % ", ".join(sorted(extra)))
+        items = _as_id_list(id)
+        if len(items) != 1:
+            raise ValueError("aligned= takes exactly one id")
+        item = items[0]
+        id_map = {}
+        if _VFB_ID_RE.match(item):
+            vfb_id = item
+        else:
+            site = self.xref_db
+            if site is None:
+                raise ValueError(
+                    "Instance '%s' (project %d) has no VFB skid cross-"
+                    "references, so an aligned copy cannot be looked up "
+                    "from a skeleton id — none exists without a VFB record"
+                    % (self.instance, self.project_id))
+            mapping = skids_to_vfb_ids([item], site)
+            if item not in mapping:
+                raise ValueError(
+                    "Skeleton id %s has no VFB record on '%s', so VFB "
+                    "holds no aligned copy" % (item, self.instance))
+            vfb_id = mapping[item]
+            id_map = {vfb_id: item}
+
+        registrations = list_aligned_templates(vfb_id)
+        if not registrations:
+            raise ValueError("VFB holds no template-registered image for %s"
+                             % vfb_id)
+        available = ", ".join("%s (%s)" % (r["template"], r["label"])
+                              for r in registrations)
+        generic = (choice is True or
+                   str(choice).strip().lower() in ("true", "1", "yes", "vfb"))
+        if generic:
+            if len(registrations) > 1:
+                raise ValueError(
+                    "%s is registered to %d templates — pass "
+                    "aligned=<template short_form or label>: %s"
+                    % (vfb_id, len(registrations), available))
+            chosen = registrations
+        else:
+            want = str(choice).strip().lower()
+            chosen = [r for r in registrations
+                      if r["template"].lower() == want
+                      or (r.get("label") or "").lower() == want]
+            if not chosen:
+                raise ValueError(
+                    "%s is not registered to template '%s'. Available: %s"
+                    % (vfb_id, choice, available))
+        reg = chosen[0]
+        url = reg["folder"].rstrip("/") + "/volume.swc"
+        if url.startswith("http://"):
+            url = "https://" + url[len("http://"):]
+        resp = _http_session().get(url, timeout=CATMAID_TIMEOUT)
+        if resp.status_code != 200:
+            raise ValueError(
+                "No aligned SWC for %s in %s (%s returned %d) — the image "
+                "may not have a skeleton representation"
+                % (vfb_id, reg["label"], url, resp.status_code))
+        swc_text = resp.content.decode("utf-8", "replace")
+        if raw:
+            return swc_text
+        return {
+            "instance": self.instance,
+            "project_id": self.project_id,
+            "command": "swc",
+            "aligned": True,
+            "template": {"short_form": reg["template"], "label": reg["label"]},
+            "xref_db": self.xref_db,
+            "id_map": id_map,
+            "unmatched": [],
+            "reverse_map": {v: k for k, v in id_map.items()},
+            "notes": ["aligned SWC served from the VFB image store: %s" % url],
+            "result": swc_text,
+        }
+
+    def _swc_alignments(self, id, raw=False, extra=None):
+        """The spaces an SWC download is available in for one neuron.
+
+        Always lists the original EM-space skeleton (CATMAID); adds one row
+        per VFB template registration, each checked for an actual
+        ``volume.swc`` in the image store. Every row carries the ``aligned=``
+        value that fetches it through the ``swc`` command.
+        """
+        if extra:
+            raise ValueError(
+                "swc_alignments takes only id= — unexpected: %s"
+                % ", ".join(sorted(extra)))
+        items = _as_id_list(id)
+        if len(items) != 1:
+            raise ValueError("swc_alignments takes exactly one id")
+        item = items[0]
+        site = self.xref_db
+        id_map, notes = {}, []
+        if _VFB_ID_RE.match(item):
+            vfb_id = item
+            skid = None
+            if site:
+                mapping = vfb_ids_to_skids([item], site)
+                skid = mapping.get(item)
+            if skid is None:
+                raise ValueError(
+                    "%s has no skeleton id on '%s' (project %d)%s"
+                    % (item, self.instance, self.project_id,
+                       "" if site else " — this instance has no VFB skid "
+                                       "cross-references"))
+            id_map = {vfb_id: skid}
+        else:
+            skid = item
+            vfb_id = None
+            if site:
+                vfb_id = skids_to_vfb_ids([item], site).get(item)
+                if vfb_id:
+                    id_map = {vfb_id: skid}
+            if vfb_id is None:
+                notes.append(
+                    "skeleton id %s has no VFB record on '%s', so only the "
+                    "original EM-space skeleton is available" % (skid,
+                                                                 self.instance))
+
+        alignments = [{
+            "aligned": "original",
+            "label": "%s EM space (CATMAID original)" % self._meta.get(
+                "name", self.instance),
+            "source": "catmaid",
+            "swc_available": True,
+        }]
+        if vfb_id:
+            for reg in list_aligned_templates(vfb_id):
+                url = reg["folder"].rstrip("/") + "/volume.swc"
+                if url.startswith("http://"):
+                    url = "https://" + url[len("http://"):]
+                try:
+                    ok = _http_session().head(
+                        url, allow_redirects=True,
+                        timeout=CATMAID_TIMEOUT).status_code == 200
+                except Exception as exc:
+                    log.warning("HEAD %s failed: %s", url, exc)
+                    ok = False
+                alignments.append({
+                    "aligned": reg["template"],
+                    "label": reg.get("label"),
+                    "template": reg["template"],
+                    "source": "vfb",
+                    "swc_available": ok,
+                    "url": url,
+                })
+
+        result = {"id": item, "vfb_id": vfb_id, "skid": skid,
+                  "alignments": alignments}
+        if raw:
+            return result
+        envelope = {
+            "instance": self.instance,
+            "project_id": self.project_id,
+            "command": "swc_alignments",
+            "xref_db": site,
+            "id_map": id_map,
+            "unmatched": [],
+            "reverse_map": {v: k for k, v in id_map.items()},
+            "result": result,
+        }
+        if notes:
+            envelope["notes"] = notes
+        return envelope
+
+    # -- HTTP ---------------------------------------------------------------
+
+    def _request(self, method, path, params):
+        url = self.base_url + path
+        headers = {}
+        if self.token:
+            headers["X-Authorization"] = "Token %s" % self.token
+        if method == "GET":
+            resp = _http_session().get(url, params=params, headers=headers,
+                                       timeout=CATMAID_TIMEOUT)
+        else:
+            resp = _http_session().post(url, data=params, headers=headers,
+                                        timeout=CATMAID_TIMEOUT)
+        content_type = resp.headers.get("Content-Type", "")
+        if resp.status_code >= 400:
+            detail = resp.text[:500]
+            raise RuntimeError(
+                "CATMAID %s %s returned %d: %s"
+                % (method, path, resp.status_code, detail))
+        if "json" in content_type:
+            body = resp.json()
+            # CATMAID reports many errors as 200 + {"error": ...}
+            if isinstance(body, dict) and body.get("error"):
+                raise RuntimeError("CATMAID error on %s %s: %s"
+                                   % (method, path, body.get("error")))
+            return {"body": body, "kind": "json"}
+        return {"body": resp.content, "kind": "raw"}
+
+    # -- the pass-through ---------------------------------------------------
+
+    def call(self, command, raw=False, **kwargs):
+        """Run one registry command; see the module docstring for the
+        envelope. ``raw=True`` returns the untouched CATMAID response."""
+        spec = CATMAID_COMMANDS.get(command)
+        if spec is None:
+            raise ValueError(
+                "Unknown CATMAID command '%s'. Available: %s"
+                % (command, ", ".join(sorted(CATMAID_COMMANDS))))
+
+        # Local commands are answered by VFBquery itself, not CATMAID.
+        if command == "swc_alignments":
+            return self._swc_alignments(kwargs.pop("id", None), raw=raw,
+                                        extra=kwargs)
+
+        # swc has one option CATMAID itself cannot serve: VFB's template-
+        # registered copy of the skeleton, downloaded from the VFB image
+        # store instead of CATMAID. aligned= names the template (short_form
+        # or label); "original"/"catmaid"/falsy mean the CATMAID original.
+        if command == "swc":
+            choice = kwargs.pop("aligned", None)
+            if choice is not None and not (
+                    choice is False or
+                    str(choice).strip().lower() in
+                    ("", "false", "0", "no", "original", "catmaid")):
+                return self._aligned_swc(kwargs.pop("id", None), choice,
+                                         raw=raw, extra=kwargs)
+
+        path_slots = {"project_id": self.project_id}
+        params = {}
+        id_map, unmatched = {}, []
+        notes = []
+
+        # Declared skid-list parameters.
+        for public, wire in (spec.get("id_params") or {}).items():
+            if public not in kwargs:
+                raise ValueError("Command '%s' requires '%s' (skeleton ids "
+                                 "and/or VFB ids)" % (command, public))
+            skids, this_map, this_unmatched = self.resolve_ids(
+                kwargs.pop(public))
+            id_map.update(this_map)
+            unmatched.extend(this_unmatched)
+            if not skids:
+                raise ValueError(
+                    "No usable ids for '%s' after conversion (unmatched: %s)"
+                    % (public, ", ".join(unmatched) or "none"))
+            for i, skid in enumerate(skids):
+                params[wire.format(i=i)] = skid
+
+        # Single path id (skid or neuron id).
+        if spec.get("id_path"):
+            if "id" not in kwargs:
+                raise ValueError("Command '%s' requires 'id' (one skeleton id "
+                                 "or VFB id)" % command)
+            skids, this_map, this_unmatched = self.resolve_ids(
+                kwargs.pop("id"))
+            id_map.update(this_map)
+            unmatched.extend(this_unmatched)
+            if len(skids) != 1:
+                raise ValueError(
+                    "Command '%s' takes exactly one resolvable id "
+                    "(got %d usable; unmatched: %s)"
+                    % (command, len(skids), ", ".join(unmatched) or "none"))
+            slot, kind = spec["id_path"]["slot"], spec["id_path"]["kind"]
+            value = skids[0]
+            if kind == "neuron_id":
+                neuron_id = self._skid_to_neuron_id(value)
+                notes.append("neuron_id %s derived from skeleton id %s"
+                             % (neuron_id, value))
+                value = neuron_id
+            path_slots[slot] = value
+
+        # Other raw path parameters (treenode_id etc.).
+        for name in spec.get("path_params") or []:
+            if name not in kwargs:
+                raise ValueError("Command '%s' requires '%s'" % (command, name))
+            path_slots[name] = str(kwargs.pop(name)).strip("/")
+
+        # Everything else goes to CATMAID verbatim.
+        for key, value in kwargs.items():
+            if isinstance(value, bool):
+                value = "true" if value else "false"
+            params[str(key)] = value
+
+        path = spec["path"].format(**path_slots)
+        response = self._request(spec["method"], path, params)
+
+        returns = spec.get("returns", "json")
+        if response["kind"] == "json":
+            result = response["body"]
+        elif returns == "bytes":
+            result = base64.b64encode(response["body"]).decode("ascii")
+            notes.append("binary response, base64-encoded")
+        else:
+            result = response["body"].decode("utf-8", "replace")
+
+        if raw:
+            return result
+
+        envelope = {
+            "instance": self.instance,
+            "project_id": self.project_id,
+            "command": command,
+            "xref_db": self.xref_db,
+            "id_map": id_map,
+            "unmatched": sorted(set(unmatched)),
+            "result": result,
+        }
+        if notes:
+            envelope["notes"] = notes
+
+        # Reverse-map skid-shaped keys found in the result back to VFB ids —
+        # cheap (one batched KB query) and only where xrefs exist at all.
+        site = self.xref_db
+        if site and isinstance(result, (dict, list)):
+            found = set()
+            _collect_skid_like_keys(result, found, MAX_REVERSE_LOOKUP)
+            found.update(id_map.values())
+            if found:
+                try:
+                    envelope["reverse_map"] = skids_to_vfb_ids(found, site)
+                except Exception as exc:
+                    log.warning("reverse skid->VFB lookup failed: %s", exc)
+                    envelope["reverse_map"] = {
+                        v: k for k, v in id_map.items()}
+        elif id_map:
+            envelope["reverse_map"] = {v: k for k, v in id_map.items()}
+        return envelope
+
+    def __getattr__(self, name):
+        if name in CATMAID_COMMANDS:
+            def _bound(**kwargs):
+                return self.call(name, **kwargs)
+            _bound.__name__ = name
+            _bound.__doc__ = CATMAID_COMMANDS[name]["doc"]
+            return _bound
+        raise AttributeError(
+            "%r object has no attribute %r (not a CATMAID command either — "
+            "see .commands())" % (type(self).__name__, name))
+
+    def __dir__(self):
+        return sorted(set(list(super().__dir__()) + list(CATMAID_COMMANDS)))
+
+
+# ---------------------------------------------------------------------------
+# Public module-level API
+# ---------------------------------------------------------------------------
+
+def catmaid(instance, project=None):
+    """A :class:`CatmaidInstance` for one hosted instance (e.g. ``'fafb'``)."""
+    return CatmaidInstance(instance, project=project)
+
+
+def list_catmaid_instances():
+    """Metadata for every VFB-hosted CATMAID instance.
+
+    Straight from catmaid.json, with each project annotated with the KB
+    xref site (``vfb_xref_db``) used for VFB id <-> skid conversion, or
+    None where the KB holds no skid xrefs (skid-only access).
+    """
+    config = get_catmaid_config()
+    try:
+        site_map = _xref_site_map()
+    except Exception:
+        site_map = dict(_STATIC_XREF_SITES)
+    instances = []
+    for inst in config.get("instances", []):
+        entry = {k: inst.get(k) for k in
+                 ("id", "name", "description", "url", "api_documentation",
+                  "more_information", "api_token")}
+        entry["projects"] = [
+            dict(p, vfb_xref_db=site_map.get((inst.get("id"), int(p["id"]))))
+            for p in inst.get("projects", [])]
+        instances.append(entry)
+    return {
+        "name": config.get("name"),
+        "description": config.get("description"),
+        "source": CATMAID_JSON_URL,
+        "homepage": config.get("homepage"),
+        "citation": config.get("citation"),
+        "authentication": config.get("authentication"),
+        "instances": instances,
+    }
+
+
+def list_catmaid_commands():
+    """{command: {method, path, doc, takes}} for the whole registry."""
+    out = {}
+    for name, spec in sorted(CATMAID_COMMANDS.items()):
+        takes = list(spec.get("id_params") or [])
+        if spec.get("id_path"):
+            takes.append("id")
+        takes.extend(spec.get("path_params") or [])
+        out[name] = {"method": spec["method"], "path": spec["path"],
+                     "doc": spec["doc"], "takes_ids": takes,
+                     "returns": spec.get("returns", "json"),
+                     "local": bool(spec.get("local"))}
+    return out
+
+
+def run_catmaid_command(instance, command, project=None, raw=False, **kwargs):
+    """Module-level convenience (and the ha_api worker entry point)."""
+    return CatmaidInstance(instance, project=project).call(
+        command, raw=raw, **kwargs)
diff --git a/src/vfbquery/ha_api.py b/src/vfbquery/ha_api.py
index bd262ec..6081aac 100644
--- a/src/vfbquery/ha_api.py
+++ b/src/vfbquery/ha_api.py
@@ -25,6 +25,9 @@
     GET /search?query=<free_text>                      # canonical website search
     GET /facets[?contains=<text>]                      # type names /search accepts
     GET /xref?id=<VFB id> | ?accession=<external id>[&db=<site>]
+    GET /catmaid                                       # hosted CATMAID instances
+    GET /catmaid/{instance}                            # metadata + commands
+    GET /catmaid/{instance}/{command}?ids=<skids/VFB ids>[&project=][&raw=true]
     GET /health
     GET /status          — queue depth, cache stats & worker utilisation
 
@@ -489,7 +492,13 @@ def snapshot(self):
     "/resolve_combination", "/find_combo_publications",
     "/list_connectome_datasets", "/query_connectivity",
     "/search", "/facets", "/xref", "/combine", "/get_hierarchy",
+    "/catmaid",
 })
+
+#: Prefixes under which *dynamic* routes are allowed. Exact-match only would
+#: 404 every /catmaid/{instance}/{command} path; anything under a prefix here
+#: is passed to normal routing (unknown sub-paths still 404 via the router).
+ALLOWED_PATH_PREFIXES = ("/catmaid/",)
 # /facets belongs here because /search's four type parameters 400 with
 # "did you mean" suggestions on an unrecognised name and point the caller at
 # /facets for the full list — an allowlist that 404s the endpoint the error
@@ -546,7 +555,7 @@ def unreachable_routes(app):
     registered = set()
     for route in app.router.routes():
         path = getattr(route.resource, "canonical", None)
-        if path:
+        if path and not path.startswith(ALLOWED_PATH_PREFIXES):
             registered.add(path)
     return sorted(registered - set(ALLOWED_PATHS))
 
@@ -568,7 +577,9 @@ async def security_middleware(request, handler):
     block and is passed to normal routing -- an unknown path still 404s via the
     router, but is not counted or logged as a probe.
     """
-    if request.path not in ALLOWED_PATHS and not _is_trusted_remote(request.remote):
+    if (request.path not in ALLOWED_PATHS
+            and not request.path.startswith(ALLOWED_PATH_PREFIXES)
+            and not _is_trusted_remote(request.remote)):
         probes = request.app.get("_scanner_probes")
         if probes is None:
             probes = {"count": 0}
@@ -1725,7 +1736,7 @@ async def _compute():
 
 
 async def _dispatch_to_pool(request, cache_key, worker_fn, *args, post_fn=None,
-                            known_params=None):
+                            known_params=None, client_error_types=()):
     """Shared dispatch logic for new endpoints — cache, coalesce, queue, run.
 
     If *post_fn* is given it is called on the result **after** cache
@@ -1736,6 +1747,13 @@ async def _dispatch_to_pool(request, cache_key, worker_fn, *args, post_fn=None,
     worker needed) because graph builders are lightweight CPU work plus
     a single Neo4j batch lookup.
 
+    *client_error_types* is a tuple of exception types that mean "the caller
+    asked for something that does not exist or does not parse" — they come
+    back as a 400 with the exception's message rather than a 500. Only
+    handlers whose workers validate input (the /catmaid family, whose
+    instance list lives on a remote server the event loop should not fetch
+    synchronously) set this; everything else keeps the old contract.
+
     *known_params* is the set of query-string keys the calling handler reads.
     Anything else the caller sent is reported back as a warning rather than
     ignored in silence — see :func:`_unknown_param_warnings`. It is applied
@@ -1774,6 +1792,8 @@ def finish(result):
         except Overloaded as exc:
             return _overloaded_response(exc)
         except Exception as exc:
+            if client_error_types and isinstance(exc, client_error_types):
+                return web.json_response({"error": str(exc)}, status=400)
             return _failure_response("Query failed", exc, cache_key, coalesced=True)
 
     tracker = request.app["tracker"]
@@ -1794,6 +1814,8 @@ def finish(result):
     except Overloaded as exc:
         return _overloaded_response(exc)
     except Exception as exc:
+        if client_error_types and isinstance(exc, client_error_types):
+            return web.json_response({"error": str(exc)}, status=400)
         return _failure_response("Query failed", exc, cache_key)
 
 
@@ -3707,6 +3729,125 @@ async def handle_combine(request):
     return web.json_response(_cap_result_rows(result))
 
 
+# ---------------------------------------------------------------------------
+# /catmaid — pass-through to the VFB-hosted CATMAID instances
+#
+# GET /catmaid                        list hosted instances (catmaid.json + KB
+#                                     xref sites)
+# GET /catmaid/{instance}             one instance's metadata + the command
+#                                     registry
+# GET /catmaid/{instance}/{command}   run one read-only CATMAID command;
+#                                     `ids` (comma-separated) accepts CATMAID
+#                                     skeleton ids and/or VFB ids, `project`
+#                                     picks a project id, `raw=true` returns
+#                                     the untouched CATMAID response, and any
+#                                     other parameter is forwarded to CATMAID
+#                                     verbatim.
+#
+# The heavy lifting — instance registry, id conversion through the KB xrefs,
+# neuron-id bridging and the response envelope — lives in catmaid_client;
+# these handlers only parse the URL and ride the shared dispatch machinery
+# (cache, coalescing, queue, compute budget). The envelope is what gets
+# cached; `raw` is applied by post_fn so both views share one cache entry.
+# ---------------------------------------------------------------------------
+
+#: Query-string keys the /catmaid/{instance}/{command} handler itself reads.
+#: Everything else is forwarded to CATMAID, so no unknown-param warnings.
+_CATMAID_CONTROL_PARAMS = frozenset({"project", "raw"})
+
+_CATMAID_INSTANCE_RE = re.compile(r"^[a-z0-9][a-z0-9.\-]{0,31}$")
+_CATMAID_COMMAND_RE = re.compile(r"^[a-z0-9_]{1,64}$")
+
+
+def _run_catmaid_instances():
+    """Execute list_catmaid_instances in a worker process."""
+    from .catmaid_client import list_catmaid_instances
+    return list_catmaid_instances()
+
+
+def _run_catmaid_instance(instance):
+    """One instance's metadata plus the command registry, in a worker."""
+    from .catmaid_client import list_catmaid_instances, list_catmaid_commands
+    listing = list_catmaid_instances()
+    match = [i for i in listing["instances"] if i["id"] == instance]
+    if not match:
+        raise ValueError(
+            "Unknown CATMAID instance '%s'. Hosted instances: %s"
+            % (instance,
+               ", ".join(sorted(i["id"] for i in listing["instances"]))))
+    return {"instance": match[0], "commands": list_catmaid_commands(),
+            "usage": "/catmaid/%s/{command}?ids=<skids and/or VFB ids>"
+                     "[&project=<pid>][&raw=true]" % instance}
+
+
+def _run_catmaid_command(instance, command, project, params):
+    """Execute one CATMAID pass-through command in a worker process."""
+    from .catmaid_client import run_catmaid_command
+    return run_catmaid_command(instance, command, project=project, **params)
+
+
+def _catmaid_raw_view(result):
+    """post_fn for raw=true: unwrap the envelope, leave anything else alone."""
+    if isinstance(result, dict) and "result" in result and "command" in result:
+        return result["result"]
+    return result
+
+
+async def handle_catmaid_instances(request):
+    """GET /catmaid — the hosted CATMAID instances and their metadata."""
+    return await _dispatch_to_pool(
+        request, "catmaid|instances", _run_catmaid_instances,
+        known_params=frozenset(), client_error_types=(ValueError,))
+
+
+async def handle_catmaid_instance(request):
+    """GET /catmaid/{instance} — instance metadata + available commands."""
+    instance = request.match_info["instance"].lower()
+    if not _CATMAID_INSTANCE_RE.match(instance):
+        return web.json_response({"error": "Malformed instance name"},
+                                 status=400)
+    return await _dispatch_to_pool(
+        request, "catmaid|instance|%s" % instance, _run_catmaid_instance,
+        instance, known_params=frozenset(), client_error_types=(ValueError,))
+
+
+async def handle_catmaid_command(request):
+    """GET /catmaid/{instance}/{command} — run one pass-through command."""
+    instance = request.match_info["instance"].lower()
+    command = request.match_info["command"].lower()
+    if not _CATMAID_INSTANCE_RE.match(instance):
+        return web.json_response({"error": "Malformed instance name"},
+                                 status=400)
+    if not _CATMAID_COMMAND_RE.match(command):
+        return web.json_response({"error": "Malformed command name"},
+                                 status=400)
+
+    project = (request.query.get("project") or "").strip() or None
+    if project is not None and not project.isdigit():
+        return web.json_response({"error": "project must be a numeric "
+                                           "CATMAID project id"}, status=400)
+    raw = (request.query.get("raw") or "").strip().lower() in (
+        "1", "true", "yes")
+
+    params = {}
+    for key in set(request.query.keys()) - _CATMAID_CONTROL_PARAMS:
+        values = request.query.getall(key)
+        params[key] = values[0] if len(values) == 1 else values
+
+    # `raw` is deliberately NOT part of the key: both views come from the
+    # one cached envelope, unwrapped by post_fn after cache retrieval.
+    cache_key = "catmaid|%s|%s|%s|%s" % (
+        instance, command, project or "",
+        json.dumps(sorted(params.items()), separators=(",", ":")))
+
+    return await _dispatch_to_pool(
+        request, cache_key, _run_catmaid_command,
+        instance, command, project, params,
+        post_fn=_catmaid_raw_view if raw else None,
+        known_params=None,  # everything unrecognised is forwarded to CATMAID
+        client_error_types=(ValueError,))
+
+
 # ---------------------------------------------------------------------------
 # Application factory
 # ---------------------------------------------------------------------------
@@ -3772,6 +3913,14 @@ def create_app(max_workers=None, max_concurrent=None, max_queue_depth=None,
     # Set algebra over query results
     app.router.add_get("/combine", handle_combine)
 
+    # CATMAID pass-through
+    app.router.add_get("/catmaid", handle_catmaid_instances)
+    app.router.add_get("/catmaid/", handle_catmaid_instances)
+    app.router.add_get("/catmaid/{instance}", handle_catmaid_instance)
+    app.router.add_get("/catmaid/{instance}/", handle_catmaid_instance)
+    app.router.add_get("/catmaid/{instance}/{command}", handle_catmaid_command)
+    app.router.add_get("/catmaid/{instance}/{command}/", handle_catmaid_command)
+
     _warn_unreachable_routes(app)
 
     # Store config for /status and handlers