From 27aa05f54e82f2a985e5a611ef93e244c04db08c Mon Sep 17 00:00:00 2001 From: Paul Hunkin Date: Thu, 22 Oct 2026 12:23:56 +0100 Subject: [PATCH] feat(community): add Jsonify dataset rows tool (#7253) * feat(community): add Jsonify dataset rows tool Adds deerflow.community.jsonify.tools:dataset_rows_tool, a read-only tool that returns rows from maintained Jsonify datasets. Jsonify pipelines run on a schedule and repair themselves when a source changes, so the agent reads a current table instead of scraping. - operator configures a named map of dataset -> serve URL; the model picks by name and never sees the URL - serve URLs must be https without userinfo and pass validate_public_http_url before any request; the URL is redacted from error text - paging with has_more/next_offset, version pinned after the first page, limit clamped to the configured max_rows, system columns stripped - config.example.yaml entry under tools; unit tests cover paging, clamping, unknown dataset, unconfigured tool, http/userinfo rejection, private address block and error redaction Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01QrmeK98ZKoxZNiUiz8kVeX * fix(community): harden Jsonify tool paging, logging and error handling - Stop paging on an empty page or a next_offset that does not advance - Log config validation failures by field location and error type only - Validate page shape; malformed or non-JSON pages return a clean error - Transport errors report the exception type, never the URL - Add truncated flag when rows were clipped or more rows remain - Tests for each case Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QrmeK98ZKoxZNiUiz8kVeX --------- Co-authored-by: Claude Fable 5.1 --- .../deerflow/community/jsonify/__init__.py | 0 .../deerflow/community/jsonify/tools.py | 200 +++++++++++++ backend/tests/test_jsonify_tools.py | 276 ++++++++++++++++++ config.example.yaml | 13 + 4 files changed, 489 insertions(+) create mode 100644 backend/packages/harness/deerflow/community/jsonify/__init__.py create mode 100644 backend/packages/harness/deerflow/community/jsonify/tools.py create mode 100644 backend/tests/test_jsonify_tools.py diff --git a/backend/packages/harness/deerflow/community/jsonify/__init__.py b/backend/packages/harness/deerflow/community/jsonify/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/backend/packages/harness/deerflow/community/jsonify/tools.py b/backend/packages/harness/deerflow/community/jsonify/tools.py new file mode 100644 index 0000000000..ed71674b5a --- /dev/null +++ b/backend/packages/harness/deerflow/community/jsonify/tools.py @@ -0,0 +1,200 @@ +"""Read rows from maintained Jsonify datasets. + +Jsonify pipelines run on a schedule, repair themselves when a source changes, +and publish the result as a dataset. Each dataset has a serve URL that returns +its latest rows as JSON with paging, column selection, exact-match filters and +version pinning. The URL is the credential: it is configured by the operator, +never passed by the model, and scrubbed from any error text. + +Configure one entry per deployment with a named map of datasets:: + + - name: jsonify_dataset_rows + group: data + use: deerflow.community.jsonify.tools:dataset_rows_tool + datasets: + llm_api_pricing: $JSONIFY_LLM_PRICING_URL + max_rows: 100 + timeout: 30 +""" + +from __future__ import annotations + +import asyncio +import json +import logging +from typing import Any + +import httpx +from langchain_core.tools import StructuredTool +from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, ValidationError, field_validator + +from deerflow.community.provider_credentials import redact_secret, reject_url_userinfo +from deerflow.community.url_safety import validate_public_http_url +from deerflow.config import get_app_config + +logger = logging.getLogger(__name__) + +_TOOL_NAME = "jsonify_dataset_rows" +_PAGE_SIZE = 1000 +_MAX_FILTERS = 10 +_SYSTEM_COLUMNS = ("_parent_row_id", "_source_key") + + +class _JsonifySettings(BaseModel): + """Validated provider settings stored on the tool entry.""" + + model_config = ConfigDict(validate_default=True) + + datasets: dict[str, AnyHttpUrl] = Field(min_length=1, max_length=50) + max_rows: int = Field(default=100, ge=1, le=10_000) + timeout: float = Field(default=30, gt=0, le=600) + + @field_validator("datasets") + @classmethod + def _check_urls(cls, value: dict[str, AnyHttpUrl]) -> dict[str, AnyHttpUrl]: + for name, url in value.items(): + if not name.strip(): + raise ValueError("dataset names must not be blank") + reject_url_userinfo(url) + if url.scheme != "https": + raise ValueError(f"dataset '{name}': serve URL must use https") + return value + + +class _RowsArgs(BaseModel): + dataset: str = Field(description="Name of the configured dataset to read.") + filters: dict[str, str] | None = Field( + default=None, + description="Exact, case-sensitive column matches combined with AND, for example {'provider': 'Anthropic'}. Up to 10.", + ) + columns: list[str] | None = Field(default=None, description="Column names to return. Omit for all columns.") + limit: int = Field(default=50, ge=1, description="Maximum number of rows to return.") + + +def _load_settings() -> _JsonifySettings | str: + config = get_app_config().get_tool_config(_TOOL_NAME) + extra = dict(config.model_extra or {}) if config is not None else {} + try: + return _JsonifySettings(**extra) + except ValidationError as exc: + # Log only where and why validation failed; the input values are serve URLs, which are credentials. + problems = "; ".join(f"{'.'.join(str(part) for part in err['loc']) or ''}: {err['type']}" for err in exc.errors()) + logger.warning("Invalid %s configuration: %s", _TOOL_NAME, problems) + return f"Error: {_TOOL_NAME} is not configured. Add a 'datasets' map of name to serve URL in config.yaml." + + +def _build_params( + *, + limit: int, + offset: int, + columns: list[str] | None, + filters: dict[str, str] | None, + version: int | None, +) -> dict[str, Any]: + params: dict[str, Any] = {"limit": limit, "offset": offset} + if columns: + params["columns"] = ",".join(c.strip() for c in columns if c.strip()) + if version is not None: + params["version"] = version + for column, value in (filters or {}).items(): + params[f"filter[{column}]"] = value + return params + + +async def _read_rows(settings: _JsonifySettings, args: _RowsArgs) -> dict[str, Any]: + serve_url = str(settings.datasets[args.dataset]) + url_error = await asyncio.to_thread(validate_public_http_url, serve_url, action="read") + if url_error: + raise ValueError(url_error) + + max_rows = min(args.limit, settings.max_rows) + rows: list[dict[str, Any]] = [] + meta: dict[str, Any] = {} + offset = 0 + version: int | None = None + truncated = False + async with httpx.AsyncClient(timeout=settings.timeout) as client: + while True: + limit = min(_PAGE_SIZE, max_rows - len(rows)) + response = await client.get( + serve_url, + params=_build_params(limit=limit, offset=offset, columns=args.columns, filters=args.filters, version=version), + ) + response.raise_for_status() + page = _parse_page(response) + if not meta: + meta = { + "dataset": args.dataset, + "dataset_name": page.get("dataset_name"), + "table": page.get("table"), + "version": page.get("version"), + "updated_at": page.get("updated_at"), + } + if version is None: + version = page.get("version") + page_rows = [{k: v for k, v in row.items() if k not in _SYSTEM_COLUMNS} for row in page["rows"]] + rows.extend(page_rows) + has_more = bool(page.get("has_more")) + next_offset = page.get("next_offset") + if len(rows) >= max_rows: + truncated = has_more or len(rows) > max_rows + break + # Stop on an empty page or a cursor that does not advance, so a misbehaving endpoint cannot spin the loop. + if not has_more or not page_rows or not isinstance(next_offset, int) or next_offset <= offset: + break + offset = next_offset + + del rows[max_rows:] + return {**meta, "row_count": len(rows), "truncated": truncated, "rows": rows} + + +def _parse_page(response: httpx.Response) -> dict[str, Any]: + """Decode one serve response and check it has the expected shape.""" + try: + page = response.json() + except ValueError as exc: + raise ValueError("Jsonify returned a response that is not JSON") from exc + rows = page.get("rows") if isinstance(page, dict) else None + if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows): + raise ValueError("Jsonify returned an unexpected response shape") + return page + + +async def _dataset_rows( + dataset: str, + filters: dict[str, str] | None = None, + columns: list[str] | None = None, + limit: int = 50, +) -> str: + settings = _load_settings() + if isinstance(settings, str): + return settings + if dataset not in settings.datasets: + available = ", ".join(sorted(settings.datasets)) + return f"Error: Unknown dataset '{dataset}'. Available datasets: {available}" + if filters and len(filters) > _MAX_FILTERS: + return f"Error: At most {_MAX_FILTERS} filters are supported" + + serve_url = str(settings.datasets[dataset]) + try: + args = _RowsArgs(dataset=dataset, filters=filters, columns=columns, limit=limit) + result = await _read_rows(settings, args) + except (ValidationError, ValueError, TypeError, AttributeError) as exc: + return redact_secret(f"Error: {exc}", serve_url) + except httpx.HTTPStatusError as exc: + return redact_secret(f"Error: Jsonify returned HTTP {exc.response.status_code}", serve_url) + except httpx.HTTPError as exc: + return f"Error: Could not reach Jsonify: {type(exc).__name__}" + return json.dumps(result, ensure_ascii=False) + + +dataset_rows_tool = StructuredTool.from_function( + coroutine=_dataset_rows, + name=_TOOL_NAME, + description=( + "Read rows from a maintained Jsonify dataset. Rows are the latest published output of a scheduled, " + "self-repairing pipeline, so they are current without scraping. Pass the dataset name, optional exact-match " + "filters, optional columns and a row limit. Returns JSON with dataset metadata, version, updated_at, rows and a truncated flag that is true when more rows matched than were returned." + ), + args_schema=_RowsArgs, +) diff --git a/backend/tests/test_jsonify_tools.py b/backend/tests/test_jsonify_tools.py new file mode 100644 index 0000000000..9ca11c0b4c --- /dev/null +++ b/backend/tests/test_jsonify_tools.py @@ -0,0 +1,276 @@ +"""Unit tests for the Jsonify community tool.""" + +import ipaddress +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +SERVE_URL = "https://factory.jsonify.com/api/serve/abc123" +PAGE_ONE = { + "dataset": "abc123", + "dataset_name": "LLM API pricing", + "table": "models", + "version": 3, + "updated_at": 1791979608, + "row_count": 2, + "has_more": True, + "next_offset": 2, + "rows": [ + {"_row_id": "r1", "_parent_row_id": "p1", "_source_key": "anthropic", "provider": "Anthropic", "model": "a"}, + {"_row_id": "r2", "_parent_row_id": "p2", "_source_key": "anthropic", "provider": "Anthropic", "model": "b"}, + ], +} +PAGE_TWO = {**PAGE_ONE, "row_count": 1, "has_more": False, "next_offset": None, "rows": [{"_row_id": "r3", "provider": "Anthropic", "model": "c"}]} + + +@pytest.fixture +def public_dns(): + with patch( + "deerflow.community.url_safety.resolve_host_addresses", + return_value=[ipaddress.ip_address("93.184.216.34")], + ): + yield + + +def _config(extra: dict | None): + tool_config = MagicMock() + tool_config.model_extra = extra + app_config = MagicMock() + app_config.get_tool_config.return_value = tool_config + return app_config + + +def _response(payload: dict, status: int = 200) -> httpx.Response: + return httpx.Response(status, json=payload, request=httpx.Request("GET", SERVE_URL)) + + +def _client_with(responses: list[httpx.Response]) -> MagicMock: + client = MagicMock() + client.get = AsyncMock(side_effect=responses) + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=None) + return client + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +async def test_reads_pages_strips_system_columns_and_pins_version(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = _client_with([_response(PAGE_ONE), _response(PAGE_TWO)]) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = json.loads(await dataset_rows_tool.ainvoke({"dataset": "pricing", "filters": {"provider": "Anthropic"}, "columns": ["provider", "model"], "limit": 10})) + + assert result["dataset_name"] == "LLM API pricing" + assert result["row_count"] == 3 + assert result["rows"][0] == {"_row_id": "r1", "provider": "Anthropic", "model": "a"} + first = client.get.call_args_list[0].kwargs["params"] + second = client.get.call_args_list[1].kwargs["params"] + assert first == {"limit": 10, "offset": 0, "columns": "provider,model", "filter[provider]": "Anthropic"} + assert second["offset"] == 2 + assert second["version"] == 3 + assert second["limit"] == 8 + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +async def test_limit_is_capped_by_configured_max_rows(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = _client_with([_response(PAGE_ONE)]) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}, "max_rows": 1})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = json.loads(await dataset_rows_tool.ainvoke({"dataset": "pricing", "limit": 500})) + + assert client.get.call_args.kwargs["params"]["limit"] == 1 + assert result["row_count"] == 1 + + +@pytest.mark.anyio +async def test_unknown_dataset_lists_available_names(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + with patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL, "releases": SERVE_URL + "0"}})): + result = await dataset_rows_tool.ainvoke({"dataset": "nope"}) + + assert result == "Error: Unknown dataset 'nope'. Available datasets: pricing, releases" + + +@pytest.mark.anyio +async def test_unconfigured_tool_returns_error_without_request(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config(None)), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient") as client_cls, + ): + result = await dataset_rows_tool.ainvoke({"dataset": "pricing"}) + + assert result.startswith("Error: jsonify_dataset_rows is not configured") + client_cls.assert_not_called() + + +@pytest.mark.anyio +async def test_rejects_http_serve_url_and_userinfo(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + for bad in ("http://factory.jsonify.com/api/serve/abc", "https://user:pw@factory.jsonify.com/api/serve/abc"): + with patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": bad}})): + result = await dataset_rows_tool.ainvoke({"dataset": "pricing"}) + assert result.startswith("Error: jsonify_dataset_rows is not configured") + + +@pytest.mark.anyio +async def test_private_address_blocked_before_request(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"meta": "https://169.254.169.254/latest/meta-data"}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient") as client_cls, + ): + result = await dataset_rows_tool.ainvoke({"dataset": "meta"}) + + assert result.startswith("Error:") + client_cls.assert_not_called() + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +async def test_http_error_is_redacted(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = _client_with([_response({"error": "nope"}, status=404)]) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = await dataset_rows_tool.ainvoke({"dataset": "pricing"}) + + assert result == "Error: Jsonify returned HTTP 404" + assert SERVE_URL not in result + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +async def test_truncated_flag_when_more_rows_match(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = _client_with([_response(PAGE_ONE)]) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = json.loads(await dataset_rows_tool.ainvoke({"dataset": "pricing", "limit": 2})) + + assert result["row_count"] == 2 + assert result["truncated"] is True + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +async def test_complete_read_is_not_truncated(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = _client_with([_response(PAGE_TWO)]) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = json.loads(await dataset_rows_tool.ainvoke({"dataset": "pricing"})) + + assert result["truncated"] is False + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +@pytest.mark.parametrize( + "page", + [ + {**PAGE_ONE, "next_offset": 0}, + {**PAGE_ONE, "next_offset": "2"}, + {**PAGE_ONE, "rows": []}, + ], +) +async def test_stops_when_cursor_does_not_advance_or_page_is_empty(page): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = _client_with([_response(page), _response(PAGE_TWO)]) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = json.loads(await dataset_rows_tool.ainvoke({"dataset": "pricing", "limit": 100})) + + assert client.get.call_count == 1 + assert result["row_count"] == len(page["rows"]) + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +@pytest.mark.parametrize("payload", [["not", "a", "page"], {"rows": "nope"}, {"rows": ["r1"]}, {"has_more": False}]) +async def test_malformed_page_returns_redacted_error(payload): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = MagicMock() + client.get = AsyncMock(return_value=httpx.Response(200, json=payload, request=httpx.Request("GET", SERVE_URL))) + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=None) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = await dataset_rows_tool.ainvoke({"dataset": "pricing"}) + + assert result == "Error: Jsonify returned an unexpected response shape" + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +async def test_non_json_page_returns_error(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = _client_with([httpx.Response(200, text="", request=httpx.Request("GET", SERVE_URL))]) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = await dataset_rows_tool.ainvoke({"dataset": "pricing"}) + + assert result == "Error: Jsonify returned a response that is not JSON" + + +def test_invalid_config_warning_omits_serve_url(caplog): + from deerflow.community.jsonify.tools import _load_settings + + secret = "http://factory.jsonify.com/api/serve/supersecret" + with patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": secret}})): + result = _load_settings() + + assert isinstance(result, str) + assert "supersecret" not in caplog.text + assert "datasets: value_error" in caplog.text + + +@pytest.mark.anyio +@pytest.mark.usefixtures("public_dns") +async def test_transport_error_omits_url(): + from deerflow.community.jsonify.tools import dataset_rows_tool + + client = MagicMock() + client.get = AsyncMock(side_effect=httpx.ConnectError(f"failed to connect to {SERVE_URL}")) + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=None) + with ( + patch("deerflow.community.jsonify.tools.get_app_config", return_value=_config({"datasets": {"pricing": SERVE_URL}})), + patch("deerflow.community.jsonify.tools.httpx.AsyncClient", return_value=client), + ): + result = await dataset_rows_tool.ainvoke({"dataset": "pricing"}) + + assert result == "Error: Could not reach Jsonify: ConnectError" diff --git a/config.example.yaml b/config.example.yaml index ee20a4273f..8fbba4cb69 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -888,6 +888,19 @@ tools: # group: knowledge # use: deerflow.community.ragflow.tools:list_knowledge_bases_tool + # Jsonify dataset rows (read-only). Jsonify pipelines run on a schedule, + # repair themselves when a source changes, and publish the result as a + # dataset. Each dataset has a serve URL (Integrations > REST API in Jsonify) + # that returns its latest rows as JSON. The URL is the credential: keep it in + # an env var, never in the prompt. Name each dataset; the model picks by name. + # - name: jsonify_dataset_rows + # group: data + # use: deerflow.community.jsonify.tools:dataset_rows_tool + # datasets: + # llm_api_pricing: $JSONIFY_LLM_PRICING_URL # https://factory.jsonify.com/api/serve/ + # max_rows: 100 # Cap per call; the model's limit is clamped to this + # timeout: 30 + # LightRAG knowledge retrieval (read-only). Alternative provider for the same # knowledge_search tool; uncomment this entry instead of the RAGFlow one — # duplicate names keep the first entry, so configure exactly one. Requires