From 992fcd572d754d83659572b1c2e0456be505b2ac Mon Sep 17 00:00:00 2001 From: Zio Gabber <78922322+Gabrymi93@users.noreply.github.com> Date: Mon, 17 Aug 2026 18:21:22 +0100 Subject: [PATCH] =?UTF-8?q?refactor(mcp):=2016=20=E2=86=92=205=20tool=20ag?= =?UTF-8?q?gregati=20(dataset,=20query,=20pipeline,=20source,=20contract)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sostituisce i 16 tool flat con 5 tool CLI-like, ognuno con param action che fa da dispatch. Riduce la complessita' per gli agenti AI da 16 nomi da ricordare a 5 pattern uniformi. - toolkit_dataset: find, overview, status, preflight, schema-diff - toolkit_query: run (SQL), preview (URL CSV/TSV) - toolkit_pipeline: contract, runs, registry_list/show, graph - toolkit_source: probe, ckan, links, sparql - toolkit_contract: backward compat Test: 34/34, ruff clean. --- tests/test_mcp_server.py | 400 +++++++++++++++------------ toolkit/mcp/server.py | 582 +++++++++++++++++++-------------------- 2 files changed, 505 insertions(+), 477 deletions(-) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index a495b58..9515514 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -16,193 +16,292 @@ def test_mcp_server_registers_expected_tools() -> None: tools = asyncio.run(mcp_server.mcp.list_tools()) tool_names = {tool.name for tool in tools} assert tool_names == { - "toolkit_list_runs", - "toolkit_schema_diff", - "toolkit_layer", - "toolkit_status", + "toolkit_dataset", + "toolkit_query", + "toolkit_pipeline", + "toolkit_source", "toolkit_contract", - "toolkit_probe_url", - "toolkit_ckan_package_show", - "toolkit_html_extract_links", - "toolkit_sparql_query", - "toolkit_preview_url", - "toolkit_preflight", - "toolkit_find", - "toolkit_dataset_overview", - "toolkit_registry_list", - "toolkit_registry_show", - "toolkit_graph", } -def test_toolkit_contract_structure() -> None: - """toolkit_contract returns stable structure with all required keys.""" - result = mcp_server.toolkit_contract(layer="all") - assert "version" in result - assert "pipeline" in result +# --------------------------------------------------------------------------- +# toolkit_dataset +# --------------------------------------------------------------------------- + + +def test_toolkit_dataset_find(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_find(**kwargs): + return {"datasets": [], "total_count": 0} + + monkeypatch.setattr(mcp_server, "find_impl", fake_find) + result = mcp_server.toolkit_dataset(action="find", query="terna") + assert "datasets" in result + + +def test_toolkit_dataset_overview(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_overview(**kwargs): + return {"slug": kwargs.get("slug"), "columns": []} + + monkeypatch.setattr(mcp_server, "dataset_overview_impl", fake_overview) + result = mcp_server.toolkit_dataset(action="overview", slug="terna_electricity_by_source") + assert result["slug"] == "terna_electricity_by_source" + + +def test_toolkit_dataset_overview_missing_slug() -> None: + result = mcp_server.toolkit_dataset(action="overview") + assert "error" in result + + +def test_toolkit_dataset_status(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_status(config, **kwargs): + return {"config": str(config)} + + monkeypatch.setattr(mcp_server, "dataset_status_impl", fake_status) + result = mcp_server.toolkit_dataset(action="status", config_path="dataset.yml") + assert result["config"] == "dataset.yml" + + +def test_toolkit_dataset_status_missing_config() -> None: + result = mcp_server.toolkit_dataset(action="status") + assert "error" in result + + +def test_toolkit_dataset_preflight(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_preflight(config, *, years_arg=None): + return {"config": str(config), "status": "passed"} + + monkeypatch.setattr("toolkit.domain.preflight.run_preflight", fake_preflight) + result = mcp_server.toolkit_dataset(action="preflight", config_path="dataset.yml") + assert result["status"] == "passed" + + +def test_toolkit_dataset_schema_diff(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_diff(config): + return {"config": str(config), "diff": []} + + monkeypatch.setattr(mcp_server, "schema_diff_impl", fake_diff) + result = mcp_server.toolkit_dataset(action="schema-diff", config_path="dataset.yml") + assert result["diff"] == [] + + +def test_toolkit_dataset_invalid_action() -> None: + result = mcp_server.toolkit_dataset(action="bogus") + assert result["error"] == "invalid_action" + + +# --------------------------------------------------------------------------- +# toolkit_query +# --------------------------------------------------------------------------- + + +def test_toolkit_query_run(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_layer(**kwargs): + return {"columns": ["anno"], "rows": [{"anno": 2024}]} + + monkeypatch.setattr(mcp_server, "layer_query_impl", fake_layer) + result = mcp_server.toolkit_query( + action="run", datasets=["terna"], sql="SELECT * FROM terna LIMIT 1" + ) + assert "columns" in result + + +def test_toolkit_query_run_missing_sql() -> None: + result = mcp_server.toolkit_query(action="run", datasets=["terna"]) + assert "error" in result + + +def test_toolkit_query_preview(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_preview(url, **kwargs): + return {"url": url, "columns": []} + + monkeypatch.setattr(mcp_server, "preview_url_impl", fake_preview) + result = mcp_server.toolkit_query(action="preview", url="https://example.it/data.csv") + assert result["url"] == "https://example.it/data.csv" + + +def test_toolkit_query_invalid_action() -> None: + result = mcp_server.toolkit_query(action="bogus") + assert result["error"] == "invalid_action" + + +# --------------------------------------------------------------------------- +# toolkit_pipeline +# --------------------------------------------------------------------------- + + +def test_toolkit_pipeline_contract() -> None: + result = mcp_server.toolkit_pipeline(action="contract", layer="clean") + assert result["layer"] == "clean" + assert "sql_source" in result + + +def test_toolkit_pipeline_contract_all() -> None: + result = mcp_server.toolkit_pipeline(action="contract") assert "clean" in result assert "mart" in result - assert "constants" in result - assert "tldr" in result - # Clean contract has macros with warning rules - clean = result["clean"] - assert clean["sql_source"]["view"] == "raw_input" - assert len(clean["macros"]) >= 8 - italian_macro = [m for m in clean["macros"] if m["name"] == "normalize_italian_number"] - assert len(italian_macro) == 1 - assert "warning" in italian_macro[0] - # Layer-specific queries - # Layer-specific queries - raw_only = mcp_server.toolkit_contract(layer="raw") - assert raw_only["layer"] == "raw" - assert "source_types" in raw_only - assert any(s["type"] == "http_file" for s in raw_only["source_types"]) +def test_toolkit_pipeline_runs(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_runs(config, *args, **kwargs): + return {"runs": []} - clean_only = mcp_server.toolkit_contract(layer="clean") - assert clean_only["layer"] == "clean" - assert "sql_source" in clean_only - assert clean_only["sql_source"]["view"] == "raw_input" + monkeypatch.setattr(mcp_server, "list_runs_impl", fake_runs) + result = mcp_server.toolkit_pipeline(action="runs", config_path="dataset.yml") + assert "runs" in result - mart_only = mcp_server.toolkit_contract(layer="mart") - assert mart_only["layer"] == "mart" - assert "sql_source" in mart_only - assert mart_only["sql_source"]["view"] == "clean_input" +def test_toolkit_pipeline_registry_list(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_list(): + return {"repos": []} -def test_tool_returns_payload_on_success(monkeypatch: pytest.MonkeyPatch) -> None: - """Through a real tool implementation, guard passes payload through unchanged.""" - monkeypatch.setattr(mcp_server, "probe_url_impl", lambda url, timeout: {"ok": True}) - result = mcp_server.toolkit_probe_url("https://example.gov.it", timeout=15) - assert result == {"ok": True} + monkeypatch.setattr(mcp_server, "registry_list_impl", fake_list) + result = mcp_server.toolkit_pipeline(action="registry_list") + assert "repos" in result + + +def test_toolkit_pipeline_registry_show(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_show(repo, artifact, slug=None): + return {"repo": repo, "artifact": artifact} + + monkeypatch.setattr(mcp_server, "registry_show_impl", fake_show) + result = mcp_server.toolkit_pipeline( + action="registry_show", repo="eurostat", artifact="datasets" + ) + assert result["repo"] == "eurostat" + + +def test_toolkit_pipeline_graph(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_graph(**kwargs): + return {"nodes": [], "edges": []} + + monkeypatch.setattr(mcp_server, "graph_impl", fake_graph) + result = mcp_server.toolkit_pipeline(action="graph", by_domain="appalti") + assert "nodes" in result + + +def test_toolkit_pipeline_invalid_action() -> None: + result = mcp_server.toolkit_pipeline(action="bogus") + assert result["error"] == "invalid_action" # --------------------------------------------------------------------------- -# Scout tool contract tests +# toolkit_source # --------------------------------------------------------------------------- -def test_toolkit_probe_url_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None: - calls: dict = {} - - def fake_impl(url: str, timeout: int) -> dict: - calls.update(url=url, timeout=timeout) +def test_toolkit_source_probe(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_probe(url, timeout): return {"status_code": 200} - monkeypatch.setattr(mcp_server, "probe_url_impl", fake_impl) - result = mcp_server.toolkit_probe_url("https://example.gov.it", timeout=30) - assert result == {"status_code": 200} - assert calls == {"url": "https://example.gov.it", "timeout": 30} + monkeypatch.setattr(mcp_server, "probe_url_impl", fake_probe) + result = mcp_server.toolkit_source(action="probe", url="https://example.gov.it") + assert result["status_code"] == 200 -def test_toolkit_probe_url_with_routed(monkeypatch: pytest.MonkeyPatch) -> None: - """toolkit_probe_url con routed=True usa l'implementazione routed.""" - calls: dict = {} +def test_toolkit_source_ckan(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_ckan(endpoint, package_id, timeout): + return {"title": "Test"} - def fake_impl(url: str, timeout: int) -> dict: - calls.update(url=url, timeout=timeout) - return {"source_type": "ckan"} + monkeypatch.setattr(mcp_server, "ckan_package_show_impl", fake_ckan) + result = mcp_server.toolkit_source( + action="ckan", endpoint="https://dati.gov.it", package_id="test" + ) + assert result["title"] == "Test" - monkeypatch.setattr(mcp_server, "probe_url_routed_impl", fake_impl) - result = mcp_server.toolkit_probe_url("https://dati.gov.it", timeout=15, routed=True) - assert result == {"source_type": "ckan"} - assert calls == {"url": "https://dati.gov.it", "timeout": 15} +def test_toolkit_source_links(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_links(url, timeout): + return {"total": 1} -def test_toolkit_ckan_package_show_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None: - calls: dict = {} + monkeypatch.setattr(mcp_server, "html_extract_links_impl", fake_links) + result = mcp_server.toolkit_source(action="links", url="https://example.gov.it/pagina") + assert result["total"] == 1 - def fake_impl(endpoint: str, package_id: str, timeout: int) -> dict: - calls.update(endpoint=endpoint, package_id=package_id, timeout=timeout) - return {"title": "Test dataset", "resources": []} - monkeypatch.setattr(mcp_server, "ckan_package_show_impl", fake_impl) - result = mcp_server.toolkit_ckan_package_show("https://dati.gov.it", "test-dataset", timeout=30) - assert result == {"title": "Test dataset", "resources": []} - assert calls == {"endpoint": "https://dati.gov.it", "package_id": "test-dataset", "timeout": 30} +def test_toolkit_source_sparql(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_sparql(endpoint, query, timeout, max_rows): + return {"columns": ["s"], "total_rows": 1} + monkeypatch.setattr(mcp_server, "sparql_query_impl", fake_sparql) + result = mcp_server.toolkit_source( + action="sparql", endpoint="https://e.org/sparql", query="SELECT * WHERE {?s ?p ?o}" + ) + assert result["total_rows"] == 1 -def test_toolkit_html_extract_links_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None: - calls: dict = {} - def fake_impl(url: str, timeout: int) -> dict: - calls.update(url=url, timeout=timeout) - return {"total": 2, "data_links": [{"url": "https://ex.it/data.csv"}], "groups": []} +def test_toolkit_source_invalid_action() -> None: + result = mcp_server.toolkit_source(action="bogus") + assert result["error"] == "invalid_action" - monkeypatch.setattr(mcp_server, "html_extract_links_impl", fake_impl) - result = mcp_server.toolkit_html_extract_links("https://example.gov.it/pagina", timeout=20) - assert result["total"] == 2 - assert len(result["data_links"]) == 1 - assert calls == {"url": "https://example.gov.it/pagina", "timeout": 20} +# --------------------------------------------------------------------------- +# toolkit_contract (backward compat) +# --------------------------------------------------------------------------- -def test_toolkit_sparql_query_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None: - calls: dict = {} - def fake_impl(endpoint: str, query: str, timeout: int, max_rows: int) -> dict: - calls.update(endpoint=endpoint, query=query, timeout=timeout, max_rows=max_rows) - return {"columns": ["s", "p", "o"], "total_rows": 10} +def test_toolkit_contract_structure() -> None: + result = mcp_server.toolkit_contract(layer="all") + assert "version" in result + assert "pipeline" in result + assert "clean" in result + assert "mart" in result + assert "constants" in result + assert "tldr" in result + + clean = result["clean"] + assert clean["sql_source"]["view"] == "raw_input" + assert len(clean["macros"]) >= 8 + + raw_only = mcp_server.toolkit_contract(layer="raw") + assert raw_only["layer"] == "raw" + assert "source_types" in raw_only + + clean_only = mcp_server.toolkit_contract(layer="clean") + assert clean_only["layer"] == "clean" + assert clean_only["sql_source"]["view"] == "raw_input" + + mart_only = mcp_server.toolkit_contract(layer="mart") + assert mart_only["layer"] == "mart" + assert mart_only["sql_source"]["view"] == "clean_input" + + +# --------------------------------------------------------------------------- +# Integration: tool returns payload through guard_timed +# --------------------------------------------------------------------------- - monkeypatch.setattr(mcp_server, "sparql_query_impl", fake_impl) - result = mcp_server.toolkit_sparql_query( - "https://example.org/sparql", "SELECT * WHERE {?s ?p ?o}", timeout=60, max_rows=500 - ) - assert result == {"columns": ["s", "p", "o"], "total_rows": 10} - assert calls == { - "endpoint": "https://example.org/sparql", - "query": "SELECT * WHERE {?s ?p ?o}", - "timeout": 60, - "max_rows": 500, - } +def test_tool_returns_payload_on_success(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(mcp_server, "probe_url_impl", lambda url, timeout: {"ok": True}) + result = mcp_server.toolkit_source(action="probe", url="https://example.gov.it", timeout=15) + assert result == {"ok": True} -def test_toolkit_probe_url_error_has_error_code(monkeypatch: pytest.MonkeyPatch) -> None: + +def test_toolkit_source_probe_error_has_error_code(monkeypatch: pytest.MonkeyPatch) -> None: from lab_connectors.mcp import ErrorCode as LabErrorCode def failing_impl(url: str, timeout: int) -> dict: raise ToolkitClientError("test probe error") monkeypatch.setattr(mcp_server, "probe_url_impl", failing_impl) - - payload = mcp_server.toolkit_probe_url("https://example.gov.it", timeout=15) + payload = mcp_server.toolkit_source(action="probe", url="https://example.gov.it", timeout=15) assert "error" in payload assert "message" in payload assert payload["error"] == LabErrorCode.UNEXPECTED.value -def test_toolkit_probe_url_returns_payload(monkeypatch: pytest.MonkeyPatch) -> None: - """guard_timed passes payload through unchanged for scout tools.""" - - def fake_impl(url: str, timeout: int) -> dict: - return {"status_code": 200, "content_type": "text/csv"} - - monkeypatch.setattr(mcp_server, "probe_url_impl", fake_impl) - result = mcp_server.toolkit_probe_url("https://example.gov.it/data.csv", timeout=15) - assert result == {"status_code": 200, "content_type": "text/csv"} - - -# inspect_schema e inspect_profile rimossi come tool MCP — -# coperti da toolkit_layer(mode="schema") e toolkit_status +# --------------------------------------------------------------------------- +# CSV preview (schema_ops unit tests — unchanged) +# --------------------------------------------------------------------------- def test_csv_preview_returns_profiler_aligned_fields(tmp_path: Path) -> None: - """csv_preview output must include sniff params and be compatible with profiler. - - Regression test: ensures csv_preview reuses sniff_source_file and - profile_with_read_cfg so mapping_suggestions, delim_suggested, - encoding_suggested, decimal_suggested, skip_suggested, and - robust_read_suggested are all present and consistent with profile_raw. - """ from toolkit.mcp.schema_ops import csv_preview - # Italian decimal CSV: semicolon delim, comma decimal csv_path = tmp_path / "italian.csv" csv_path.write_text("Regione;Valore\nLombardia;1.234,56\nLazio;7.890,12\n", encoding="utf-8") result = csv_preview(str(csv_path), limit=10) - # Must have profiler alignment fields assert "delim_suggested" in result assert "encoding_suggested" in result assert "decimal_suggested" in result @@ -212,12 +311,10 @@ def test_csv_preview_returns_profiler_aligned_fields(tmp_path: Path) -> None: assert result["decimal_suggested"] == "," assert result["encoding_suggested"] is not None - # mapping_suggestions must be present and valid assert "mapping_suggestions" in result mapping = result["mapping_suggestions"] assert "Regione" in mapping or "Valore" in mapping - # Basic schema fields still present assert result["path"] == str(csv_path) assert result["column_count"] == 2 assert len(result["preview"]) == 2 @@ -225,65 +322,29 @@ def test_csv_preview_returns_profiler_aligned_fields(tmp_path: Path) -> None: def test_csv_preview_ragged_csv_succeeds_with_robust_read(tmp_path: Path) -> None: - """csv_preview must succeed on ragged/IRPEF-like CSV (header < data cols). - - When profile_with_read_cfg retries with robust fallback (null_padding), - csv_preview preview/count phase must also use the robust fallback, - not the original cfg that would fail on ragged rows. - Regression test for the fix: preview phase must use robust_preset - when robust_read_suggested=True. - """ from toolkit.mcp.schema_ops import csv_preview - # Ragged CSV: header has 2 cols, data rows have 3 cols csv_path = tmp_path / "ragged.csv" csv_path.write_text("a;b\n1;2;3\n4;5;6\n", encoding="utf-8") result = csv_preview(str(csv_path), limit=10) - # Must succeed without raising ToolkitClientError assert "preview" in result assert "mapping_suggestions" in result - # robust_read_suggested must be True since ragged rows need null_padding assert result["robust_read_suggested"] is True - # Preview still returns data assert len(result["preview"]) == 2 -def test_toolkit_preflight_returns_report(monkeypatch: pytest.MonkeyPatch) -> None: - """toolkit_preflight passa config e years a run_preflight.""" - calls: dict[str, object] = {} - - def fake_preflight(config, *, years_arg=None): - calls["config"] = str(config) - calls["years_arg"] = years_arg - return {"config": str(config), "sources": [], "years": [2024], "status": "passed"} - - monkeypatch.setattr( - "toolkit.domain.preflight.run_preflight", - fake_preflight, - ) - - result = mcp_server.toolkit_preflight("dataset.yml", years="2024") - - assert result["status"] == "passed" - assert calls["config"] == "dataset.yml" - assert calls["years_arg"] == "2024" - - # --------------------------------------------------------------------------- -# mcp_sparql_query — flattening SPARQL bindings → righe MCP +# SPARQL flattening (scout_ops unit tests — unchanged) # --------------------------------------------------------------------------- def _make_fake_bindings(*rows: dict[str, str]) -> list[dict[str, dict]]: - """Costruisce bindings SPARQL finti (formato {var: {type, value}}).""" return [{k: {"type": "literal", "value": v} for k, v in row.items()} for row in rows] def test_mcp_sparql_query_flattens_bindings(monkeypatch: pytest.MonkeyPatch) -> None: - """Binding SPARQL JSON ({var: {type, value}}) → righe piatte {var: value}.""" - def _fake_sparql(_endpoint: str, _query: str, timeout: int = 60) -> list[dict[str, dict]]: return _make_fake_bindings( {"s": "http://a/1", "p": "pred1", "o": "hello"}, @@ -307,7 +368,6 @@ def _fake_sparql(_endpoint: str, _query: str, timeout: int = 60) -> list[dict[st def test_mcp_sparql_query_respects_max_rows(monkeypatch: pytest.MonkeyPatch) -> None: - """Il parametro max_rows tronca i risultati e imposta truncated=True.""" many = _make_fake_bindings(*[{"x": str(i)} for i in range(50)]) def _fake_sparql(_endpoint: str, _query: str, timeout: int = 60) -> list[dict[str, dict]]: @@ -328,8 +388,6 @@ def _fake_sparql(_endpoint: str, _query: str, timeout: int = 60) -> list[dict[st def test_mcp_sparql_query_handles_empty_bindings(monkeypatch: pytest.MonkeyPatch) -> None: - """Bindings vuoti → colonne vuote, zero righe, nessun errore.""" - def _fake_sparql(_endpoint: str, _query: str, timeout: int = 60) -> list[dict[str, dict]]: return [] @@ -346,8 +404,6 @@ def _fake_sparql(_endpoint: str, _query: str, timeout: int = 60) -> list[dict[st def test_mcp_sparql_query_handles_error(monkeypatch: pytest.MonkeyPatch) -> None: - """RuntimeError da execute_sparql → dict con error, risultati vuoti.""" - def _fake_sparql(_endpoint: str, _query: str, timeout: int = 60) -> list[dict[str, dict]]: raise RuntimeError("SPARQL endpoint unreachable") diff --git a/toolkit/mcp/server.py b/toolkit/mcp/server.py index 196d161..006defd 100644 --- a/toolkit/mcp/server.py +++ b/toolkit/mcp/server.py @@ -1,19 +1,13 @@ """Toolkit MCP server. -Espone tool read-only per ispezione della pipeline toolkit. -Include sia tool granulari (backward compat) che tool aggregati. +Espone 5 tool aggregati per ispezione dataset, query, pipeline e fonti. -Tool aggregati: -- toolkit_layer: schema/preview/profile/sql su RAW/CLEAN/MART in un tool -- toolkit_status: paths + summary + readiness + run_stats + info in un tool - -Tool catalogo (nuovi, basati sui registry.json committati): -- toolkit_find: cerca dataset pubblicati su GCS per slug/layer -- toolkit_dataset_overview: schema + conteggio + preview da slug - -Tool granulari: -- schema_diff, list_runs -- scout: probe_url, ckan, sparql, html, preview_url +Tool: +- toolkit_dataset: find, overview, status, preflight, schema-diff +- toolkit_query: run (SQL), preview (URL CSV/TSV) +- toolkit_pipeline: contract, runs, registry_list, registry_show, graph +- toolkit_source: probe, ckan, links, sparql +- toolkit_contract: contratti pipeline (backward compat) Usa ``lab_connectors.mcp`` per init standardizzato, error handling e logging. """ @@ -55,221 +49,332 @@ mcp = create_mcp_server( name="toolkit", instructions=( - "Toolkit pipeline server — ispeziona dataset, esegue preview, " + "Toolkit pipeline server — ispeziona dataset, esegue query, " "e fornisce contratti per agenti AI.\n\n" - "📌 **PRIMA di scrivere clean.sql o mart.sql**: chiama " + "5 tool aggregati:\n" + "- toolkit_dataset: find, overview, status, preflight, schema-diff\n" + "- toolkit_query: run (SQL su raw/clean/mart), preview (URL CSV/TSV)\n" + "- toolkit_pipeline: contract, runs, registry_list/show, graph\n" + "- toolkit_source: probe HTTP, CKAN, HTML links, SPARQL\n" + "- toolkit_contract: contratti pipeline (backward compat)\n\n" + "📌 PRIMA di scrivere clean.sql o mart.sql: chiama " "toolkit_contract(layer='clean') per view name (raw_input), " - "macro disponibili, regole validazione e formati numerici.\n" - "Chiama toolkit_contract(layer='mart') per la view mart (clean_input).\n\n" - "Supporta slug dataset (es. 'terna-electricity-by-source') " - "al posto del path assoluto a dataset.yml." + "macro disponibili, regole validazione e formati numerici." ), ) -@mcp.tool( - description="Confronta i segnali di schema raw (encoding, colonne, ecc.) tra gli anni configurati per un dataset.", - structured_output=True, -) -def toolkit_schema_diff(config_path: str) -> dict[str, Any]: - return guard_timed(schema_diff_impl, "toolkit_schema_diff", config_path) +# --------------------------------------------------------------------------- +# toolkit_dataset — find, overview, status, preflight, schema-diff +# --------------------------------------------------------------------------- @mcp.tool( - description="Lista run records con filtri opzionali. Ritorna record completi (non solo metadata).", + description=( + "Ispezione dataset: find, overview, status, preflight, schema-diff.\n\n" + "Actions:\n" + "- find: cerca dataset per slug/testo/source (params: query, layer, limit, source, stage, status_filter)\n" + "- overview: schema colonne + conteggio + preview (params: slug, layer, year, source, profile)\n" + "- status: stato completo dataset (params: config_path, year, since, until)\n" + "- preflight: diagnostica pre-run (params: config_path, years)\n" + "- schema-diff: confronto schema raw tra anni (params: config_path)" + ), structured_output=True, ) -def toolkit_list_runs( - config_path: str, - year: int = 0, - *, +def toolkit_dataset( + action: str, + # find + query: str = "", + layer: str | None = None, + limit: int = 15, + source: str = "all", + stage: str = "all", + status_filter: str | None = None, + # overview + slug: str | None = None, + year: int | None = None, + profile: bool = False, + # status / preflight / schema-diff + config_path: str | None = None, since: str | None = None, until: str | None = None, - status: str | None = None, - limit: int | None = None, - cross_year: bool = False, -) -> dict[str, Any]: - return guard_timed( - list_runs_impl, - "toolkit_list_runs", - config_path, - year or None, - since=since, - until=until, - status=status, - limit=limit, - cross_year=cross_year, - ) - - -@mcp.tool( - description="Preview remoto di un URL CSV/TSV: colonne, tipi, granularità. " - "HEAD + Range GET + sniff + DuckDB profile. Solo CSV/TSV.", - structured_output=True, -) -def toolkit_preview_url( - url: str, - known_encoding: str | None = None, - known_delim: str | None = None, - known_decimal: str | None = None, - known_skip: int | None = None, + years: str | None = None, ) -> dict[str, Any]: - return guard_timed( - preview_url_impl, - "toolkit_preview_url", - url, - known_encoding=known_encoding, - known_delim=known_delim, - known_decimal=known_decimal, - known_skip=known_skip, - ) + if action == "find": + return guard_timed( + find_impl, + "toolkit_dataset_find", + query=query, + layer=layer, + limit=limit, + source=source, + stage=stage, + status_filter=status_filter, + ) + if action == "overview": + if not slug: + return {"error": "missing_param", "message": "overview richiede slug"} + return guard_timed( + dataset_overview_impl, + "toolkit_dataset_overview", + slug=slug, + layer=layer or "clean", + year=year, + source=source, + profile=profile, + ) + if action == "status": + if not config_path: + return {"error": "missing_param", "message": "status richiede config_path"} + return guard_timed( + dataset_status_impl, + "toolkit_dataset_status", + config_path, + year=year or 0, + since=since, + until=until, + ) + if action == "preflight": + if not config_path: + return {"error": "missing_param", "message": "preflight richiede config_path"} + from toolkit.domain.preflight import run_preflight + + return guard_timed(run_preflight, "toolkit_dataset_preflight", config_path, years_arg=years) + if action == "schema-diff": + if not config_path: + return {"error": "missing_param", "message": "schema-diff richiede config_path"} + return guard_timed(schema_diff_impl, "toolkit_dataset_schema_diff", config_path) + return { + "error": "invalid_action", + "message": f"Azione '{action}' non valida. Usare: find, overview, status, preflight, schema-diff", + } # --------------------------------------------------------------------------- -# Aggregated tools +# toolkit_query — run SQL, preview URL # --------------------------------------------------------------------------- @mcp.tool( - description="Query unificata su RAW/CLEAN/MART: schema, preview, profilo o SQL. " - "Due modalita': config_path (pipeline locale) o datasets (catalogo GCS/workspace). " - "mode=sql funziona su tutti i layer (raw->CSV, clean/mart->parquet). " - "Per layer=mart, table specifica la tabella (es. 'mart_top_sa'). " - "Catalog mode (datasets) supporta solo mode='sql'. " - "Con dry_run=True (mode=sql) valida lo scope SQL e fa EXPLAIN senza eseguire: " - "ritorna {'valid': True, 'plan': ...} o {'valid': False, 'error': ...} " - "— utile per provare la query prima di eseguirla. " - "Esempi: mode=sql, datasets=['anac_bandi_gara', 'popolazione_istat']", + description=( + "Query dati: SQL su dataset (raw/clean/mart) o preview URL CSV/TSV.\n\n" + "Actions:\n" + "- run: esegui SQL su uno o piu' dataset\n" + " Params: datasets (list[str]), sql, layer, mode, year, limit, dry_run, config_path, mart_index, table\n" + " Catalog mode: datasets=[slug1, slug2], sql usa gli slug come tabelle\n" + " Pipeline mode: config_path, sql usa 'data' come tabella\n" + "- preview: preview remoto CSV/TSV (params: url, known_encoding, known_delim, known_decimal, known_skip)" + ), structured_output=True, ) -def toolkit_layer( - config_path: str | None = None, +def toolkit_query( + action: str, + # run params datasets: list[str] | None = None, + sql: str | None = None, layer: str = "clean", - mode: str = "schema", + mode: str = "sql", year: int = 0, limit: int = 20, - sql: str | None = None, + dry_run: bool = False, + config_path: str | None = None, mart_index: int = 0, table: str | None = None, - dry_run: bool = False, + # preview params + url: str | None = None, + known_encoding: str | None = None, + known_delim: str | None = None, + known_decimal: str | None = None, + known_skip: int | None = None, ) -> dict[str, Any]: - return guard_timed( - layer_query_impl, - "toolkit_layer", - config_path=config_path, - datasets=datasets, - layer=layer, - mode=mode, - year=year or None, - limit=limit, - sql=sql, - mart_index=mart_index, - table=table, - dry_run=dry_run, - ) + if action == "run": + if not sql: + return {"error": "missing_param", "message": "run richiede sql"} + if not datasets and not config_path: + return {"error": "missing_param", "message": "run richiede datasets o config_path"} + return guard_timed( + layer_query_impl, + "toolkit_query_run", + config_path=config_path, + datasets=datasets, + layer=layer, + mode=mode, + year=year or None, + limit=limit, + sql=sql, + mart_index=mart_index, + table=table, + dry_run=dry_run, + ) + if action == "preview": + if not url: + return {"error": "missing_param", "message": "preview richiede url"} + return guard_timed( + preview_url_impl, + "toolkit_query_preview", + url, + known_encoding=known_encoding, + known_delim=known_delim, + known_decimal=known_decimal, + known_skip=known_skip, + ) + return { + "error": "invalid_action", + "message": f"Azione '{action}' non valida. Usare: run, preview", + } + + +# --------------------------------------------------------------------------- +# toolkit_pipeline — contract, runs, registry, graph +# --------------------------------------------------------------------------- @mcp.tool( - description="Stato completo di un dataset: paths + summary + readiness + run_stats + info. " - "Aggrega inspect_paths, summary, review_readiness, run_summary e dataset_info " - "in una unica chiamata. I parametri since/until filtrano i run per finestra temporale.", + description=( + "Pipeline toolkit: contratti, run history, registry e grafo relazioni.\n\n" + "Actions:\n" + "- contract: contratti pipeline per layer (params: layer)\n" + "- runs: lista run records (params: config_path, year, since, until, status, limit, cross_year)\n" + "- registry_list: elenca artifact registry committati\n" + "- registry_show: mostra artifact registry (params: repo, artifact, slug)\n" + "- graph: mappa relazioni tra dataset (params: by_key, by_dataset, by_registry, by_domain)" + ), structured_output=True, ) -def toolkit_status( - config_path: str, +def toolkit_pipeline( + action: str, + # contract + layer: str = "all", + # runs + config_path: str | None = None, year: int = 0, - *, since: str | None = None, until: str | None = None, + status: str | None = None, + limit: int | None = None, + cross_year: bool = False, + # registry_show + repo: str | None = None, + artifact: str | None = None, + slug: str | None = None, + # graph + by_key: str = "", + by_dataset: str = "", + by_registry: str = "", + by_domain: str = "", ) -> dict[str, Any]: - return guard_timed( - dataset_status_impl, - "toolkit_status", - config_path, - year=year or None, - since=since, - until=until, - ) + if action == "contract": + from toolkit.contracts.pipeline import CONTRACTS + + if layer == "all": + return CONTRACTS + if layer in CONTRACTS: + return {"layer": layer, **CONTRACTS[layer]} + return CONTRACTS + if action == "runs": + if not config_path: + return {"error": "missing_param", "message": "runs richiede config_path"} + return guard_timed( + list_runs_impl, + "toolkit_pipeline_runs", + config_path, + year or None, + since=since, + until=until, + status=status, + limit=limit, + cross_year=cross_year, + ) + if action == "registry_list": + return guard_timed(registry_list_impl, "toolkit_pipeline_registry_list") + if action == "registry_show": + if not repo or not artifact: + return {"error": "missing_param", "message": "registry_show richiede repo e artifact"} + return guard_timed( + registry_show_impl, "toolkit_pipeline_registry_show", repo, artifact, slug + ) + if action == "graph": + return guard_timed( + graph_impl, + "toolkit_pipeline_graph", + by_key=by_key, + by_dataset=by_dataset, + by_registry=by_registry, + by_domain=by_domain, + ) + return { + "error": "invalid_action", + "message": f"Azione '{action}' non valida. Usare: contract, runs, registry_list, registry_show, graph", + } # --------------------------------------------------------------------------- -# Catalog tools (basati sui registry.json committati) +# toolkit_source — probe, ckan, links, sparql # --------------------------------------------------------------------------- @mcp.tool( - description="Cerca dataset per slug, source, layer (clean/mart) o testo. " - "La query matcha anche la semantica dei cataloghi committati del workspace " - "(description, tags, category, nomi colonne) — vedi matched_columns/meta_match. " - "source='gcs' = pubblicati (dai registry.json committati), " - "source='workspace' = in sviluppo (da dataset.yml + parquet locali), " - "source='all' (default) = unione. " - "Filtri aggiuntivi: stage (candidates/support), status_filter (SUCCESS/FAILED/DRY_RUN), " - "metric_only (solo dataset con colonne metric dal catalogo). " - "Restituisce slug, layer, anni, file count, size, run_status, flag source + " - "semantica quando il catalogo committato è presente.", - structured_output=True, -) -def toolkit_find( - query: str = "", - layer: str | None = None, - limit: int = 15, - source: str = "all", - stage: str = "all", - status_filter: str | None = None, -) -> dict[str, Any]: - return guard_timed( - find_impl, - "toolkit_find", - query=query, - layer=layer, - limit=limit, - source=source, - stage=stage, - status_filter=status_filter, - ) - - -@mcp.tool( - description="Overview di un dataset: schema colonne (DESCRIBE DuckDB), " - "conteggio righe e preview dati. " - "Con profile=True (on-demand, default False) profila anche i valori delle " - "colonne dimensionali (cardinalità, null, top-N) — utile prima di scrivere " - "una query. " - "Parametro source='gcs' (solo pubblicati), 'workspace' (solo sviluppo), " - "'all' (default) — entrambi, preferisce locale.", + description=( + "Fonti dati esterne: probe HTTP, CKAN, HTML links, SPARQL.\n\n" + "Actions:\n" + "- probe: reachability HTTP (params: url, timeout, routed)\n" + "- ckan: fetch dataset CKAN (params: endpoint, package_id, timeout)\n" + "- links: estrai link dati da pagina HTML (params: url, timeout)\n" + "- sparql: query SPARQL SELECT (params: endpoint, query, timeout, max_rows)" + ), structured_output=True, ) -def toolkit_dataset_overview( - slug: str, - layer: str = "clean", - year: int | None = None, - source: str = "all", - profile: bool = False, +def toolkit_source( + action: str, + url: str | None = None, + endpoint: str | None = None, + package_id: str | None = None, + query: str | None = None, + timeout: int = 30, + routed: bool = False, + max_rows: int = 500, ) -> dict[str, Any]: - return guard_timed( - dataset_overview_impl, - "toolkit_dataset_overview", - slug=slug, - layer=layer, - year=year, - source=source, - profile=profile, - ) + if action == "probe": + if not url: + return {"error": "missing_param", "message": "probe richiede url"} + impl = probe_url_routed_impl if routed else probe_url_impl + name = "toolkit_source_probe" + if routed: + return guard_timed(impl, f"{name}_routed", url, timeout) + return guard_timed(impl, name, url, timeout) + if action == "ckan": + if not endpoint or not package_id: + return {"error": "missing_param", "message": "ckan richiede endpoint e package_id"} + return guard_timed( + ckan_package_show_impl, "toolkit_source_ckan", endpoint, package_id, timeout + ) + if action == "links": + if not url: + return {"error": "missing_param", "message": "links richiede url"} + return guard_timed(html_extract_links_impl, "toolkit_source_links", url, timeout) + if action == "sparql": + if not endpoint or not query: + return {"error": "missing_param", "message": "sparql richiede endpoint e query"} + return guard_timed( + sparql_query_impl, "toolkit_source_sparql", endpoint, query, timeout, max_rows + ) + return { + "error": "invalid_action", + "message": f"Azione '{action}' non valida. Usare: probe, ckan, links, sparql", + } # --------------------------------------------------------------------------- -# Pipeline contracts (AI agent interface) +# toolkit_contract — backward compat # --------------------------------------------------------------------------- @mcp.tool( - description="Restituisce i contratti di pipeline del toolkit in formato " - "strutturato. Usalo PRIMA di scrivere dataset.yml, clean.sql o mart.sql " - "per conoscere tipi fonte raw, view names (raw_input, clean_input), " - "macro disponibili, regole di validazione, e formati numerici italiani. " - "Parametro layer='raw' | 'clean' | 'mart' | 'all' (default).", + description=( + "Contratti pipeline toolkit. Usalo PRIMA di scrivere clean.sql o mart.sql " + "per conoscere view names (raw_input, clean_input), macro, regole validazione. " + "Parametro layer='raw' | 'clean' | 'mart' | 'all' (default)." + ), structured_output=True, ) def toolkit_contract(layer: str = "all") -> dict[str, Any]: @@ -282,138 +387,5 @@ def toolkit_contract(layer: str = "all") -> dict[str, Any]: return CONTRACTS -# --------------------------------------------------------------------------- -# Validate config -# --------------------------------------------------------------------------- - - -@mcp.tool( - description="Pre-flight check per un dataset: valida config, verifica " - "raggiungibilita' fonti, e per CSV produce quality score PA. " - "Non esegue la pipeline — solo diagnostica preventiva.", - structured_output=True, -) -def toolkit_preflight(config_path: str, years: str | None = None) -> dict[str, Any]: - from toolkit.domain.preflight import run_preflight - - return guard_timed(run_preflight, "toolkit_preflight", config_path, years_arg=years) - - -# --------------------------------------------------------------------------- -# Scout tools -# --------------------------------------------------------------------------- - - -@mcp.tool( - description="Probe HTTP: reachability, status code, content-type. " - "HEAD + GET Range. Nessun body scaricato. " - "Con routed=True attiva routing automatico (rileva CKAN, SDMX, HTML, file diretto).", - structured_output=True, -) -def toolkit_probe_url(url: str, timeout: int = 15, routed: bool = False) -> dict[str, Any]: - impl = probe_url_routed_impl if routed else probe_url_impl - name = "toolkit_probe_url" - if routed: - return guard_timed(impl, f"{name}_routed", url, timeout) - return guard_timed(impl, name, url, timeout) - - -@mcp.tool( - description="Fetch di un dataset CKAN via API package_show. " - "Restituisce metadati, risorse, organization, tags, formato e DataStore availability.", - structured_output=True, -) -def toolkit_ckan_package_show( - endpoint: str, - package_id: str, - timeout: int = 30, -) -> dict[str, Any]: - return guard_timed( - ckan_package_show_impl, "toolkit_ckan_package_show", endpoint, package_id, timeout - ) - - -@mcp.tool( - description="Estrae link a file dati (CSV, JSON, XLSX, ZIP, XML) da una pagina HTML. " - "Scarica la pagina, analizza i link, e restituisce URL trovati raggruppati per formato.", - structured_output=True, -) -def toolkit_html_extract_links(url: str, timeout: int = 20) -> dict[str, Any]: - return guard_timed(html_extract_links_impl, "toolkit_html_extract_links", url, timeout) - - -@mcp.tool( - description="Esegue una query SPARQL SELECT su un endpoint pubblico. " - "Restituisce risultati in formato tabellare (lista di righe con colonne). " - "Supporta qualsiasi endpoint HTTPS SPARQL.", - structured_output=True, -) -def toolkit_sparql_query( - endpoint: str, query: str, timeout: int = 60, max_rows: int = 500 -) -> dict[str, Any]: - return guard_timed( - sparql_query_impl, "toolkit_sparql_query", endpoint, query, timeout, max_rows - ) - - -@mcp.tool( - description=( - "Elenca gli artifact registry committati nei repo del workspace " - "(registry.json unico fusion). " - "Ogni repo con registry/ viene elencato con i suoi artifact, " - "dimensione e conteggio entries per sezione " - "(datasets, marts, signals, codelists, entities). " - "Usa toolkit_registry_show per il contenuto." - ), - structured_output=True, -) -def toolkit_registry_list() -> dict[str, Any]: - return guard_timed(registry_list_impl, "toolkit_registry_list") - - -@mcp.tool( - description=( - "Mostra un artifact registry committato di un repo del workspace " - "(es. repo='eurostat', artifact='datasets'). Sezioni: datasets, marts, " - "signals, codelists, entities (o 'registry' per l'intero payload). " - "Con slug filtra un'entry: dataset slug (datasets), mart slug in formato " - "{dataset}__{mart} (marts), id (signals), codelist name (codelists). " - "Il catalogo semantico contiene columns (role, semantic_type), " - "description, period, location, mart_refs e il blocco run." - ), - structured_output=True, -) -def toolkit_registry_show(repo: str, artifact: str, slug: str | None = None) -> dict[str, Any]: - return guard_timed(registry_show_impl, "toolkit_registry_show", repo, artifact, slug) - - -@mcp.tool( - description=( - "Mostra la mappa delle relazioni tra dataset del Lab (cross-repo). " - "Ogni dataset si collega a un'entità del mondo reale (Comune, Provincia, " - "Ente, Gara, ...) tramite un tipo semantico (municipality_code, fiscal_code, ...). " - "I bridge collegano entità tra loro (es. CIG → Comune via anac_bandi_gara). " - "Aggrega i registry di tutti i repo del workspace (fusion ADR). " - "Filtri: by_key (tipo semantico), by_dataset (slug), by_registry (entità), " - "by_domain (appalti/enti/territorio/giustizia/scuola/progetti/economia)." - ), - structured_output=True, -) -def toolkit_graph( - by_key: str = "", - by_dataset: str = "", - by_registry: str = "", - by_domain: str = "", -) -> dict[str, Any]: - return guard_timed( - graph_impl, - "toolkit_graph", - by_key=by_key, - by_dataset=by_dataset, - by_registry=by_registry, - by_domain=by_domain, - ) - - if __name__ == "__main__": mcp.run()