feat: add get_territorial_codes tool with DuckDB backend

Lookup REF_AREA codes by level, name, region, province, capoluogo.
9142 entries (comuni, province, regioni, ripartizioni) stored in
resources/istat_lookup.duckdb. Includes build script to regenerate
from ISTAT sources. 12 unit tests, skill docs updated.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: aborruso

LOG.md

@@ -0,0 +1,10 @@
+# Log
+
+## 2026-03-29
+
+- Add `get_territorial_codes` tool (8th tool) with DuckDB backend
+- Store territorial data in `resources/istat_lookup.duckdb` (reusable for future lookup tables)
+- Include `resources/build_territorial_subdivisions.py` script to rebuild the DB from ISTAT sources
+- Add `duckdb` dependency to `pyproject.toml`
+- 12 unit tests for territorial codes (level, name, region, province, capoluogo filters)
+- Update skill docs with territorial codes workflow note

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "tenacity>=8.2.0",
     "pydantic>=2.0.0",
     "python-dotenv>=1.0.0",
+    "duckdb>=1.0.0",
 ]
 
 [project.optional-dependencies]

resources/build_territorial_subdivisions.py

@@ -0,0 +1,235 @@
+"""Build resources/istat_lookup.duckdb (table territorial_subdivisions) from CL_ITTER107 and ISTAT data.
+
+Sources:
+- CL_ITTER107: fetched via mcp__istat__get_codelist_description (cached in cache/cache.db)
+- ISTAT municipalities CSV: https://www.istat.it/storage/codici-unita-amministrative/Elenco-comuni-italiani.csv
+- ISTAT capoluogo JSON: https://situas-servizi.istat.it/publish/reportspooljson?pfun=61&pdata=01/01/2048
+
+Output columns (table: territorial_subdivisions):
+- code: ISTAT territorial code (CL_ITTER107 for all levels; 6-digit numeric for comuni)
+- name_it: Italian name
+- level: italia | ripartizione | regione | provincia | comune
+- nuts_level: 0 | 1 | 2 | 3 | 4
+- parent_code: code of the parent territorial unit (NULL for Italia)
+- capoluogo_provincia: True if the comune is a provincial/UTS capital (NULL for non-comuni)
+- capoluogo_regione: True if the comune is a regional capital (NULL for non-comuni)
+
+Usage:
+    python3 resources/build_territorial_subdivisions.py <itter107_json> [comuni_csv]
+"""
+
+import csv
+import json
+import re
+import sys
+import tempfile
+import urllib.request
+from pathlib import Path
+
+import duckdb
+
+# NUTS 2021 codes that DON'T follow simple ITH->ITD / ITI->ITE substitution
+# Maps NUTS2021 NUTS3 code -> CL_ITTER107 code
+NUTS2021_TO_ITTER_OVERRIDE = {
+    'ITC4C': 'ITC45',  # Metro Milano -> Milano
+    'ITC4D': 'IT108',  # Monza e della Brianza
+    'ITI35': 'IT109',  # Fermo (must be before ITI->ITE rule)
+    'ITF46': 'ITF41',  # Foggia
+    'ITF47': 'ITF42',  # Bari / Citta Metropolitana
+    'ITF48': 'IT110',  # Barletta-Andria-Trani
+    'ITG2D': 'ITG25',  # Sassari
+    'ITG2E': 'ITG26',  # Nuoro
+    'ITG2F': 'ITG27',  # Cagliari
+    'ITG2G': 'ITG28',  # Oristano
+    'ITG2H': 'IT111',  # Sud Sardegna
+}
+
+# Special IT1XX province codes -> parent NUTS2 region code
+IT1XX_PARENTS = {
+    'IT108': 'ITC4',  # Monza e della Brianza -> Lombardia
+    'IT109': 'ITE3',  # Fermo -> Marche
+    'IT110': 'ITF4',  # Barletta-Andria-Trani -> Puglia
+    'IT111': 'ITG2',  # Sud Sardegna -> Sardegna
+    'IT113': 'ITG2',  # Gallura Nord-Est Sardegna -> Sardegna
+    'IT119': 'ITG2',  # Sulcis Iglesiente -> Sardegna
+}
+
+COMUNI_CSV_URL = 'https://www.istat.it/storage/codici-unita-amministrative/Elenco-comuni-italiani.csv'
+CAPOLUOGO_JSON_URL = 'https://situas-servizi.istat.it/publish/reportspooljson?pfun=61&pdata=01/01/2048'
+
+OUTPUT_PATH = Path(__file__).parent / 'istat_lookup.duckdb'
+
+
+def nuts2021_to_itter(code: str) -> str:
+    """Convert NUTS 2021 NUTS3 code to CL_ITTER107 equivalent."""
+    if code in NUTS2021_TO_ITTER_OVERRIDE:
+        return NUTS2021_TO_ITTER_OVERRIDE[code]
+    if code.startswith('ITH'):
+        return 'ITD' + code[3:]
+    if code.startswith('ITI'):
+        return 'ITE' + code[3:]
+    return code
+
+
+def load_itter107(path: str) -> dict[str, str]:
+    """Load CL_ITTER107 codelist from a JSON file (MCP tool result format)."""
+    with open(path, encoding='utf-8') as f:
+        data = json.load(f)
+    text = data[0]['text']
+    entries = re.findall(r'\{[^{}]+\}', text)
+    codes = {}
+    for e in entries:
+        cm = re.search(r'"code":\s*"([^"]+)"', e)
+        di = re.search(r'"description_it":\s*"([^"]+)"', e)
+        if cm and di:
+            codes[cm.group(1)] = di.group(1)
+    return codes
+
+
+def download_comuni_csv() -> str:
+    """Download ISTAT municipalities CSV to a temp file, return path."""
+    tmp = tempfile.NamedTemporaryFile(suffix='_comuni_istat.csv', delete=False)
+    print(f'Downloading {COMUNI_CSV_URL}...')
+    urllib.request.urlretrieve(COMUNI_CSV_URL, tmp.name)
+    utf8_path = tmp.name + '_utf8.csv'
+    with open(tmp.name, encoding='latin1') as f:
+        content = f.read()
+    with open(utf8_path, 'w', encoding='utf-8') as f:
+        f.write(content)
+    return utf8_path
+
+
+def download_capoluogo_json() -> dict[str, tuple[bool, bool]]:
+    """Download capoluogo flags from ISTAT JSON endpoint.
+
+    Returns:
+        Dict mapping 6-digit comune code -> (capoluogo_provincia, capoluogo_regione)
+    """
+    print(f'Downloading {CAPOLUOGO_JSON_URL}...')
+    with urllib.request.urlopen(CAPOLUOGO_JSON_URL) as resp:
+        data = json.loads(resp.read().decode('utf-8'))
+    result = {}
+    for rec in data.get('resultset', []):
+        code = str(rec.get('PRO_COM_T', '')).strip().zfill(6)
+        if code:
+            result[code] = (bool(rec.get('CC_UTS', 0)), bool(rec.get('CC_REG', 0)))
+    print(f'  Loaded capoluogo flags for {len(result)} comuni')
+    return result
+
+
+def build_comuni_mappings(csv_path: str) -> tuple[dict, dict]:
+    """Build comune_code->ITTER_nuts3 and storico->ITTER_nuts3 mappings."""
+    comune_to_nuts3 = {}
+    storico_to_nuts3 = {}
+    with open(csv_path, encoding='utf-8') as f:
+        reader = csv.reader(f, delimiter=';')
+        next(reader)
+        for row in reader:
+            cod6 = row[15].strip().zfill(6)
+            storico = row[2].strip().zfill(3)
+            nuts3_2021 = row[22].strip()
+            if nuts3_2021:
+                itter = nuts2021_to_itter(nuts3_2021)
+                if cod6:
+                    comune_to_nuts3[cod6] = itter
+                if storico:
+                    storico_to_nuts3[storico] = itter
+    return comune_to_nuts3, storico_to_nuts3
+
+
+def build_duckdb(codes: dict, comune_to_nuts3: dict, storico_to_nuts3: dict, capoluogo: dict[str, tuple[bool, bool]] | None = None) -> None:
+    """Build and write the DuckDB database."""
+    # Province: standard NUTS3 + letter-suffix + IT1XX
+    nuts3_codes = {k: v for k, v in codes.items() if re.match(r'^IT[A-Z][0-9]{2}$', k)}
+    letter_suffix = {k: v for k, v in codes.items() if re.match(r'^IT[A-Z][0-9][A-Z]$', k)}
+    it1xx = {k: v for k, v in codes.items() if re.match(r'^IT1[0-9]{2}$', k)}
+    all_province_codes = {**nuts3_codes, **letter_suffix, **it1xx}
+
+    rows = []
+
+    rows.append(('IT', 'Italia', 'italia', 0, None, None, None))
+
+    for k, v in {'ITC': 'Nord-ovest', 'ITD': 'Nord-est', 'ITE': 'Centro', 'ITF': 'Sud', 'ITG': 'Isole'}.items():
+        rows.append((k, v, 'ripartizione', 1, 'IT', None, None))
+
+    for k, v in codes.items():
+        if re.match(r'^IT[A-Z][0-9]$', k):
+            rows.append((k, v, 'regione', 2, k[:3], None, None))
+
+    for k, v in all_province_codes.items():
+        parent = IT1XX_PARENTS.get(k, k[:4])
+        rows.append((k, v, 'provincia', 3, parent, None, None))
+
+    no_parent = 0
+    for k, v in codes.items():
+        if re.match(r'^[0-9]{6}$', k):
+            parent = comune_to_nuts3.get(k) or storico_to_nuts3.get(k[:3])
+            if not parent:
+                no_parent += 1
+            cap_prov, cap_reg = capoluogo.get(k, (False, False)) if capoluogo else (False, False)
+            rows.append((k, v, 'comune', 4, parent, cap_prov, cap_reg))
+
+    # Remove existing db if present
+    if OUTPUT_PATH.exists():
+        OUTPUT_PATH.unlink()
+
+    conn = duckdb.connect(str(OUTPUT_PATH))
+    conn.execute('''
+        CREATE TABLE territorial_subdivisions (
+            code VARCHAR NOT NULL,
+            name_it VARCHAR NOT NULL,
+            level VARCHAR NOT NULL,
+            nuts_level TINYINT,
+            parent_code VARCHAR,
+            capoluogo_provincia BOOLEAN,
+            capoluogo_regione BOOLEAN
+        )
+    ''')
+    conn.executemany(
+        'INSERT INTO territorial_subdivisions VALUES (?, ?, ?, ?, ?, ?, ?)',
+        rows,
+    )
+    conn.execute('CREATE INDEX idx_level ON territorial_subdivisions(level)')
+    conn.execute('CREATE INDEX idx_parent ON territorial_subdivisions(parent_code)')
+    conn.execute('CREATE INDEX idx_code ON territorial_subdivisions(code)')
+    conn.close()
+
+    size = OUTPUT_PATH.stat().st_size
+    print(f'Written: {OUTPUT_PATH}')
+    print(f'Rows: {len(rows)} | Size: {size:,} bytes ({size // 1024} KB)')
+    print(f'Province: {len(all_province_codes)} | Comuni senza parent: {no_parent}')
+
+
+if __name__ == '__main__':
+    itter107_path = sys.argv[1] if len(sys.argv) > 1 else None
+    comuni_csv_path = sys.argv[2] if len(sys.argv) > 2 else None
+
+    if not itter107_path:
+        print('Usage: python3 build_territorial_subdivisions.py <itter107_json> [comuni_csv]')
+        print('  itter107_json: MCP tool result JSON for CL_ITTER107 codelist')
+        print('  comuni_csv: ISTAT municipalities CSV (latin1); downloaded if omitted')
+        sys.exit(1)
+
+    print('Loading CL_ITTER107...')
+    codes = load_itter107(itter107_path)
+    print(f'  Loaded {len(codes)} codes')
+
+    if comuni_csv_path:
+        utf8_path = comuni_csv_path + '_utf8.csv'
+        with open(comuni_csv_path, encoding='latin1') as f:
+            content = f.read()
+        with open(utf8_path, 'w', encoding='utf-8') as f:
+            f.write(content)
+        comuni_csv_path = utf8_path
+    else:
+        comuni_csv_path = download_comuni_csv()
+
+    print('Building comuni mappings...')
+    comune_to_nuts3, storico_to_nuts3 = build_comuni_mappings(comuni_csv_path)
+    print(f'  comuni: {len(comune_to_nuts3)} | storico: {len(storico_to_nuts3)}')
+
+    print('Downloading capoluogo flags...')
+    capoluogo = download_capoluogo_json()
+
+    print('Building DuckDB...')
+    build_duckdb(codes, comune_to_nuts3, storico_to_nuts3, capoluogo)

resources/istat_lookup.duckdb

@@ -6,7 +6,7 @@ description: >
   regional/provincial data, unemployment, population, GDP, agriculture, or any other
   ISTAT dataset. Guides the discover -> constraints -> data workflow step by step.
 license: MIT
-compatibility: Requires the istat MCP server to be running (provides 7 tools for ISTAT SDMX API access).
+compatibility: Requires the istat MCP server to be running (provides 8 tools for ISTAT SDMX API access).
 metadata:
   author: ondata
   version: "1.0"
@@ -34,6 +34,7 @@ Always follow this 3-step workflow:
 | 5 | `get_concepts` | Retrieve semantic definitions of SDMX concepts |
 | 6 | `get_data` | Retrieve statistical observations |
 | 7 | `get_cache_diagnostics` | Debug tool to inspect cache status |
+| 8 | `get_territorial_codes` | Lookup REF_AREA codes by level, name, region, province, or capoluogo |
 
 ## Detailed Workflow
 
@@ -265,6 +266,33 @@ Analyze employment in Italian manufacturing sectors from 2020 to 2023.
 
 ---
 
+## Territorial Codes Lookup
+
+The `get_territorial_codes` tool resolves place names and territorial levels to ISTAT REF_AREA codes. Use it when you need to build dimension filters for `REF_AREA`.
+
+**Parameters:**
+- `level`: one of `italia`, `ripartizione`, `regione`, `provincia`, `comune`
+- `name`: substring search (case-insensitive) across all levels
+- `region`: filter comuni by region name or code
+- `province`: filter comuni by province name or code
+- `capoluogo`: if `true`, return only capoluoghi di provincia
+
+**Examples:**
+
+```json
+{"level": "regione"}
+{"name": "Milano"}
+{"level": "comune", "region": "Lombardia", "capoluogo": true}
+```
+
+The data is stored in a local DuckDB database (`resources/istat_lookup.duckdb`). To rebuild it after ISTAT updates, run:
+
+```bash
+python3 resources/build_territorial_subdivisions.py <itter107_json> [comuni_csv]
+```
+
+---
+
 ## API Reference
 
 - **Base URL**: `https://esploradati.istat.it/SDMXWS/rest`

skills/istat-mcp/SKILL.md

@@ -22,6 +22,7 @@
     handle_get_constraints,
     handle_get_data,
     handle_get_structure,
+    handle_get_territorial_codes,
 )
 from .utils.tool_helpers import configure_cache_ttls
 from .utils.logging import setup_logging
@@ -222,6 +223,36 @@ async def list_tools() -> list[Tool]:
                     'properties': {},
                 },
             ),
+            Tool(
+                name='get_territorial_codes',
+                description='Get ISTAT REF_AREA territorial codes by level (italia, ripartizione, regione, provincia, comune) or place name search. Supports filtering by region, province, and capoluogo status. Contains the full territorial hierarchy (9142 entries) with parent-child relationships. Use cases: find REF_AREA codes for get_data queries (e.g. all comuni capoluogo for accident statistics, all provinces in Southern regions for olive oil production, all comuni in a specific province for demographic analysis).',
+                inputSchema={
+                    'type': 'object',
+                    'properties': {
+                        'level': {
+                            'type': 'string',
+                            'description': "Territorial level: 'italia', 'ripartizione', 'regione', 'provincia', 'comune'",
+                        },
+                        'name': {
+                            'type': 'string',
+                            'description': 'Place name to search (substring, case-insensitive)',
+                        },
+                        'region': {
+                            'type': 'string',
+                            'description': "Filter by region name or code (e.g. 'Lombardia', 'ITC4')",
+                        },
+                        'province': {
+                            'type': 'string',
+                            'description': "Filter by province name or code (e.g. 'Milano', 'ITC45')",
+                        },
+                        'capoluogo': {
+                            'type': 'boolean',
+                            'description': 'If true, return only comuni that are capoluogo di provincia',
+                            'default': False,
+                        },
+                    },
+                },
+            ),
         ]
 
     @server.call_tool()
@@ -254,6 +285,8 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[Any]:
                 result = await handle_get_data(arguments, cache_manager, api_client, blacklist)
             elif name == 'get_cache_diagnostics':
                 result = await get_cache_diagnostics_handler()
+            elif name == 'get_territorial_codes':
+                result = await handle_get_territorial_codes(arguments)
             else:
                 raise ValueError(f'Unknown tool: {name}')
 
@@ -278,5 +311,5 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[Any]:
             logger.info('=' * 80)
             raise
 
-    logger.info('MCP server configured with 7 tools')
+    logger.info('MCP server configured with 8 tools')
     return server

src/istat_mcp_server/server.py

@@ -7,6 +7,7 @@
 from .get_constraints import handle_get_constraints
 from .get_data import handle_get_data
 from .get_structure import handle_get_structure
+from .get_territorial_codes import handle_get_territorial_codes
 
 __all__ = [
     'handle_discover_dataflows',
@@ -16,4 +17,5 @@
     'handle_get_data',
     'handle_get_constraints',
     'get_cache_diagnostics_handler',
+    'handle_get_territorial_codes',
 ]