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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QrmeK98ZKoxZNiUiz8kVeX

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

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 '<root>'}: {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,
+)

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="<html>", 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"

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/<id>
+  #   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