# Global Product Risk & Recall Monitor (`atlas_digital_assistant/global-product-risk-monitor`) Actor Cross-source product-safety recall intelligence: EU Safety Gate, US openFDA, US CPSC and Health Canada normalized into one deduplicated schema with severity, provenance and change tracking. - **URL**: https://apify.com/atlas\_digital\_assistant/global-product-risk-monitor.md - **Developed by:** [Atlas Assistent](https://apify.com/atlas_digital_assistant) (community) - **Categories:** Business, E-commerce - **Stats:** 2 total users, 1 monthly users, 100.0% runs succeeded, 0 bookmarks - **User rating**: No ratings yet ## Pricing Pay per usage This Actor is paid per platform usage. The Actor is free to use, and you only pay for the Apify platform usage, which gets cheaper the higher subscription plan you have. Learn more: https://docs.apify.com/platform/actors/running/actors-in-store#pay-per-usage ## What's an Apify Actor? Actors are a software tools running on the Apify platform, for all kinds of web data extraction and automation use cases. In Batch mode, an Actor accepts a well-defined JSON input, performs an action which can take anything from a few seconds to a few hours, and optionally produces a well-defined JSON output, datasets with results, or files in key-value store. In Standby mode, an Actor provides a web server which can be used as a website, API, or an MCP server. Actors are written with capital "A". ## How to integrate an Actor? If asked about integration, you help developers integrate Actors into their projects. You adapt to their stack and deliver integrations that are safe, well-documented, and production-ready. The best way to integrate Actors is as follows. In JavaScript/TypeScript projects, use official [JavaScript/TypeScript client](https://docs.apify.com/api/client/js.md): ```bash npm install apify-client ``` In Python projects, use official [Python client library](https://docs.apify.com/api/client/python.md): ```bash pip install apify-client ``` In shell scripts, use [Apify CLI](https://docs.apify.com/cli/docs.md): ````bash # MacOS / Linux curl -fsSL https://apify.com/install-cli.sh | bash # Windows irm https://apify.com/install-cli.ps1 | iex ```bash In AI frameworks, you might use the [Apify MCP server](https://docs.apify.com/platform/integrations/mcp.md). If your project is in a different language, use the [REST API](https://docs.apify.com/api/v2.md). For usage examples, see the [API](#api) section below. For more details, see Apify documentation as [Markdown index](https://docs.apify.com/llms.txt) and [Markdown full-text](https://docs.apify.com/llms-full.txt). # README ## Global Product Risk & Recall Monitor **One normalized feed of official product-safety recalls and risk alerts from four government sources across three regions — the EU, the United States and Canada.** Built for compliance, quality, trust-&-safety and supplier-risk teams who need machine-readable recall intelligence without stitching together four different government APIs. The actor pulls from **EU Safety Gate**, **openFDA enforcement (food / drug / device)**, the **US CPSC**, and **Health Canada Recalls & Safety Alerts**, then normalizes everything into a single, stable, deduplicated schema with severity normalization, source provenance, canonical URLs, and cross-run change tracking (`firstSeen` / `lastSeen` / `changedFields`). > No search engines are scraped and no data is fabricated. Every record comes from an official public government endpoint and carries its source URL and provenance. --- ### Who is this for (ICP) & use cases - **Compliance & regulatory affairs** — monitor recalls affecting your product categories or brands across the EU, US and Canada in one feed; export to your GRC system. - **Retail / marketplace trust & safety** — detect when a product you list has been recalled and pull the offending listing quickly (electrocution, choking, fire, contamination, etc.). - **Supplier & supply-chain risk** — watch specific manufacturers / brands / countries of origin and get alerted when a supplier is implicated in a recall. - **Quality & product safety engineering** — track hazard types (e.g. `Listeria`, `electric shock`, lithium-ion battery fires) across markets to inform design and testing. - **Insurance, due diligence & research** — build datasets of recall frequency and severity by category, country and time for underwriting or analysis. - **Consumer-safety / news / watchdog apps** — feed a "latest dangerous products" surface backed by authoritative sources. Typical workflows: run **incrementally on a schedule** (e.g. every 6 hours) and route new/changed records to Slack, a webhook, a database, or a spreadsheet. --- ### Sources, coverage & attribution | Source | Region | Endpoint (machine-readable) | Coverage | |---|---|---|---| | **EU Safety Gate** (RAPEX) | EU/EEA | `api/download/weeklyReport/list|detail/xml` | Dangerous **non-food** consumer & professional products notified by EU/EEA authorities. | | **openFDA Enforcement** | US | `api.fda.gov/{food,drug,device}/enforcement.json` | FDA recall/enforcement reports for **food, drugs and medical devices**, with recall classification. | | **US CPSC** | US | `saferproducts.gov/RestWebServices/Recall` | **Consumer product** recalls (excludes food, drugs, cars, which other agencies handle). | | **Health Canada** | Canada | `recalls-rappels.canada.ca/.../HCRSAMOpenData.json` | Recalls & safety alerts for consumer products, food (CFIA), health products, medical devices and vehicles. | **Attribution.** Data © the respective authorities: European Commission (DG Justice and Consumers), U.S. Food and Drug Administration, U.S. Consumer Product Safety Commission, and the Government of Canada (Health Canada / CFIA). This actor is an independent tool and is **not affiliated with or endorsed by** any of these bodies. Always confirm against the official `canonicalUrl` before acting. #### Known source limitations - **EU Safety Gate:** the interactive portal search API is behind a bot-protection layer that blocks automation. This actor deliberately uses the **official machine-readable weekly-report XML export** (`/api/download/weeklyReport/...`, listed as an open-data distribution on data.europa.eu) instead — stable and automation-friendly. Alerts carry the weekly report's publication date rather than a per-alert timestamp. Detail URLs are restricted to HTTPS European-Commission hosts and redirects are validated. Weekly detail files are processed oldest-first; if one fails to download, the source is reported **truncated** and its cursor is not advanced past the failed week, so it is retried on the next run (up to a 520-week / ~10-year per-run ceiling). - **openFDA:** basic paging is capped at 25,000 records per query (a documented API limit). `report_date` is the **single, immutable cursor field** (`cursorDate`) — the incremental window, the keyset watermark and the client-side cursor filter all key off it, exactly the field the query paginates and sorts on (`report_date:asc` incremental, `:desc` full-mode `lookbackDays` backfill). The more meaningful `recall_initiation_date` is surfaced as `publishedDate` (both raw dates are kept in `raw`); because the cursor never keys off the initiation date, a recall reported now with an old initiation date is never skipped as "before cursor." The three sub-datasets (food/drug/device) are **independently cursored** — each advances its own partition cursor, so a failure or truncation in one neither advances over nor skips records in another; a failure in one still returns the others (reported as a partial result). Without an API key the shared rate limit applies (about 240 requests/min, 1,000/day per IP) — this actor stays well under it for normal runs. - **US CPSC:** recalls carry no severity classification, so `severity.normalized` is `unknown` (the hazard text is preserved in `riskType`/`hazard`). Queried by recall-date window. The endpoint exposes **no server-side pagination** — a single windowed request returns the complete result set for that window; if a response is unexpectedly large (≥ 9,000 items) the window is treated as possibly truncated and the cursor is held back rather than advanced over a possibly-incomplete window. - **Health Canada:** published as a single ~15 MB daily open-data snapshot; the actor downloads the whole snapshot once per run (a complete picture) and filters client-side. The publishing `Organization` is exposed as `recallingFirm` so the brand/firm filter can find it. `Recall class` (severity) is frequently blank upstream → `unknown`. If a source is temporarily unavailable, the run **continues with the remaining sources** and reports the failure per-source (see *Partial-failure handling*). A source's incremental cursor advances **only** after a successful run whose selected records were written to the dataset, and never past failed, truncated, or filtered-out regions — so a cap, outage, filter change, or write failure can never permanently skip records. --- ### Output schema (`schemaVersion` 1.1.0) Each dataset item is one normalized recall/alert: ```jsonc { "recordId": "…40-char sha1…", // deterministic: sha1(`${source}:${sourceRecordId}`) "schemaVersion": "1.1.0", "source": "eu_safety_gate", // eu_safety_gate | openfda | us_cpsc | health_canada "sourceDataset": "weekly_report", "sourceRecordId": "10099472", "regionScope": "EU", // EU | US | CA "title": "300 LED LIGHT", "productName": "String lights — 300 LED LIGHT", "brand": null, "recallingFirm": null, "categories": ["Lighting equipment"], "hazard": "The product poses a risk of electric shock…", "riskType": "Electric shock", "severity": { "normalized": "high", "raw": "Serious risk" }, // raw always preserved "recallType": "safety_alert", "reason": "…", "actions": "Withdrawal of the product from the market", "productDescription": "…", "countries": ["Hungary", "People's Republic of China"], "imageUrls": ["https://…"], "publishedDate": "2026-07-10T00:00:00.000Z", // ISO-8601 (openFDA: recall_initiation_date) "cursorDate": "2026-07-10T00:00:00.000Z", // immutable source-native date the cursor orders by (openFDA: report_date; else == publishedDate) "canonicalUrl": "https://ec.europa.eu/safety-gate-alerts/screen/webReport/alertDetail/10099472", "provenance": { "source": "eu_safety_gate", "endpoint": "https://ec.europa.eu/safety-gate-alerts/api/download/weeklyReport/detail/xml/…", "fetchedAt": "2026-07-16T09:42:47.034Z", "sourceLastUpdated": "2026-07-10T00:00:00.000Z" }, "firstSeen": "2026-07-16T09:42:47.034Z", // stable across runs (named KV store) "lastSeen": "2026-07-16T09:42:47.034Z", "changedFields": [], // e.g. ["severity","actions"] when it changes "raw": { "caseNumber": "SR/01958/26", "level": "Serious risk", "…": "…" } } ```` **Severity normalization** (raw value always kept in `severity.raw`): | Normalized | EU Safety Gate | openFDA | Health Canada | CPSC | |---|---|---|---|---| | `high` | Serious risk | Class I | Type I | — | | `medium` | Other risk | Class II | Type II | — | | `low` | — | Class III | Type III | — | | `unknown` | Risk not yet defined | (missing) | (blank) | always | *** ### Input All fields are optional; sensible defaults ship in the input schema. ```json { "sources": ["eu_safety_gate", "openfda", "us_cpsc", "health_canada"], "mode": "incremental", "maxItems": 200, "lookbackDays": 90, "recheckWindowDays": 7, "fdaDatasets": ["food", "drug", "device"], "product": "", "brand": "", "category": "", "country": "", "riskType": "", "severity": [], "dateFrom": "", "dateTo": "", "since": "", "includeRaw": true } ``` | Field | Meaning | |---|---| | `sources` | Which sources to query (defaults to all four). | | `mode` | `incremental` = only records newer than the last run (cursor stored in a named KV store); `full` = backfill over `lookbackDays`. | | `maxItems` | Per-run budget on records processed and saved, applied **after** per-source filtering and deduplication. Selection is a **balanced round-robin across sources** so every selected source is represented. In incremental mode the dataset receives only the **new/changed** records within that budget; the cursor still advances over the processed (already-known) records so the next run continues forward. | | `lookbackDays` | First-run incremental window and full-mode backfill window. | | `recheckWindowDays` | Incremental overlap (days, default 7): re-fetch & re-compare already-committed records whose `cursorDate` is within this window below the cursor, so upstream edits are surfaced. `0` disables it. The cursor is never regressed by these re-checks. | | `fdaDatasets` | Which openFDA enforcement datasets to include. | | `product`/`brand`/`category`/`country`/`riskType` | Case-insensitive substring filters. | | `severity` | Keep only these normalized levels. | | `dateFrom`/`dateTo` | Inclusive ISO date bounds on `publishedDate`. | | `since` | Override the automatic incremental cursor. | | `includeRaw` | Include the compact `raw` audit object. | #### Example: watch lithium-battery fire risks across all regions ```json { "mode": "full", "lookbackDays": 365, "riskType": "battery", "maxItems": 500 } ``` #### Example: only EU + Canada high-severity food recalls this year ```json { "sources": ["eu_safety_gate", "health_canada"], "severity": ["high"], "category": "food", "dateFrom": "2026-01-01" } ``` *** ### Modes, dedup & change tracking - **Deterministic IDs.** `recordId` is a stable `sha1(source:sourceRecordId)`; re-running never changes an item's ID, so downstream upserts are trivial. - **Deduplication.** Records are deduped by `recordId` **before** filtering, with a documented merge policy that is a pure function of the **complete raw occurrence set** for that `recordId`. The guarantee is **order-independence over that complete set** (not associativity): given all raw occurrences, `dedupeRecords` returns a **byte-identical** merged record for any input permutation — 3+ conflicting occurrences merge to the same result under any order. It is **not** a recursive/associative merge: a merged record drops the per-occurrence candidate history (individual `publishedDate`s, blank-vs-present candidates, the pre-canonical array multiset), so callers **must not** re-merge already-merged outputs — always merge the complete raw set in one `dedupeRecords` pass (the internal whole-group reducer is module-private to prevent recursive misuse). Within that pass: array fields (images, categories, countries) are **case-insensitively unioned** with a canonical casing (the lexicographically smallest variant) and a deterministic sort; scalar fields take the value from the record with the newer `publishedDate`, falling back to the next occurrence when blank (most-complete), and on an **equal** `publishedDate` the tie is broken by the larger value (never by input order); severity keeps the highest level; provenance keeps the most recently fetched (ties broken by a total order on endpoint, then source metadata). - **Incremental cursors.** State lives in a **named key-value store** (`global-product-risk-monitor-state`) that survives scheduled runs. The incremental cursor is a **keyset watermark** `(cursorDate, recordId)` **partitioned by source *and* by query configuration** (filters + FDA datasets); openFDA is partitioned further, one cursor **per food/drug/device sub-dataset**. `cursorDate` is the immutable source-native ordering date (openFDA: `report_date`; every other source: `= publishedDate`). A source is processed oldest-first and its cursor advances to the newest record that was selected and written to the dataset, and only after those writes succeed — it advances **up to but never past** a failed source, a truncated (over-cap) window, or a filtered-out region. This makes caps, source outages, filter/config changes and dataset-write failures all **retryable**: nothing is permanently skipped, and a backlog larger than `maxItems` drains forward across runs. (Overlapping runs are a separate concern — see **Concurrency** below; they share one state document and must not overlap.) - **Change tracking, recheck window & new/changed output.** The per-record snapshot yields `firstSeen` (fixed), `lastSeen` (advances every run) and `changedFields`. In **incremental** mode the dataset receives **only** genuinely new or changed records; unchanged records are still recorded in state (so the cursor advances) but are not re-emitted. To catch upstream edits to records that are **already committed** (same publication key), each incremental run additionally re-fetches and re-compares records whose `cursorDate` falls within a bounded **recheck window** below the cursor (`recheckWindowDays`, default **7**), emitting any that changed **without regressing the cursor**. Edits to records older than that window are **not** detected — widen `recheckWindowDays` (at the cost of a larger re-fetch) if you need a longer look-back. **Full** mode emits everything selected within the lookback window. - **Partial-failure handling.** Each source — and, for openFDA, each food/drug/device sub-dataset — runs independently; a failure is logged and reported in `LAST_RUN_SUMMARY` / `OUTPUT` (with `truncated` / `failed` flags) without discarding the data that succeeded. A **failed** source/sub-dataset does not advance its cursor at all; a **truncated** (but successful) one advances only to the newest fully-captured record and holds below the uncaptured region, so the remainder is retried next run. The run only fails if **every** source fails (in which case no cursor is advanced). - **Concurrency.** **All partitions — every source, sub-dataset and query configuration — share a single key-value document** (one `STATE` key in the named store). Each run does a read‑modify‑write of that **one** JSON blob, and the Apify key-value store offers **no compare-and-swap (CAS)**. So **any two overlapping runs contend on the same document, regardless of configuration** — even runs with *different* filters/FDA datasets, because their independent partition cursors still live in the same shared blob. When runs overlap, each re-reads and merges state just before persisting (furthest-advanced cursor wins, entries are unioned), which **reduces** the window for lost progress but **cannot guarantee none**: a read‑modify‑write race can still drop one run's cursor advance or cause a record to be re-emitted; the last writer can clobber a concurrent writer's partition updates entirely. A hard "no lost progress" guarantee would require CAS, which the platform does not provide; the merge is a best-effort mitigation, not a substitute for serial execution. **Operational requirement:** for the Actor account/store, keep **`maxConcurrency = 1`** and run **non-overlapping schedules** — set schedules so a run always finishes before the next starts, across *all* configurations, never just within one. Do **not** rely on different configurations being able to run concurrently. *** ### Run it via API / schedule / webhook **Apify API — start a run:** ```bash curl -X POST "https://api.apify.com/v2/acts/atlas_digital_assistant~global-product-risk-monitor/runs?token=YOUR_TOKEN" \ -H "Content-Type: application/json" \ -d '{ "mode": "incremental", "maxItems": 200 }' ``` **Get the dataset items of the last successful run:** ```bash curl "https://api.apify.com/v2/acts/atlas_digital_assistant~global-product-risk-monitor/runs/last/dataset/items?token=YOUR_TOKEN&format=json&status=SUCCEEDED" ``` **Schedule.** Use Apify **Schedules** to run `incremental` every few hours. The first run seeds `lookbackDays`; subsequent runs only return new/changed records. **Webhooks.** Add an Apify **webhook** on `ACTOR.RUN.SUCCEEDED` pointing at your endpoint (or a Make/Zapier/n8n hook) to push new records into Slack, a database or a ticketing system. Filter on `changedFields.length > 0` for "changed since last time" alerts. *** ### Compute & pricing Runtime is dominated by the Health Canada snapshot download (~15 MB) plus a handful of small JSON/XML requests. - **Illustrative, not benchmarked.** The figures below are rough order-of-magnitude expectations, **not** measured guarantees — this repo ships no automated CU/runtime benchmark, so treat them as estimates only. A typical `incremental` run with `maxItems: 200` is expected to finish in roughly a minute on the smallest (1 GB / shared-CPU) unit and to consume on the order of **~0.002–0.01 compute units (CU)**; large backfills cost more. - **Measure your own cost.** Your exact runtime, memory and CU are shown in each run's *Runs → Compute* tab — use those numbers, not these estimates, for pricing decisions. - **Store pricing model.** This actor fits a **pay-per-result** or low **monthly-rental** model; set concrete prices from your measured costs above. *** ### Privacy, legal & fair use - **Public official data only.** All records originate from public government transparency endpoints intended for redistribution; no authentication or personal accounts are used. - **No secrets** are stored in the repo, image, input or output. No API key is required for the default configuration. - **Personal data:** recall notices are about *products and firms*, not individuals. Occasional company/importer names may appear as published by the authority; treat them per the source's terms and applicable law (e.g. GDPR for EU data — the actor stores only what the authority already publishes). - **Respectful access:** requests are modest in volume, use a descriptive User-Agent, and apply retry/back-off with `Retry-After` handling. No search engines are scraped. - **Not legal/medical advice.** openFDA data in particular is unvalidated ("do not rely on openFDA to make decisions regarding medical care"). Always verify against the `canonicalUrl` before acting. Please review each source's terms: [openFDA terms](https://open.fda.gov/terms/) · [EU Safety Gate](https://ec.europa.eu/safety-gate/) · [CPSC](https://www.cpsc.gov/) · [Health Canada open data](https://open.canada.ca/). *** ### Local development ```bash npm install npm run typecheck && npm run lint && npm run build npm test # unit tests (mocked fixtures) npm run test:smoke # real-source smoke tests (network) npm run smoke:local # full end-to-end run against local storage ``` ### Changelog - **1.1.0** — Reliability & correctness overhaul. Incremental state redesigned to per-source, per-configuration **keyset cursors** that commit only after successful dataset writes and never advance past failed/truncated/filtered regions (no permanently-skipped records). Release-gate hardening: openFDA now orders/paginates/watermarks on a single immutable `cursorDate` = `report_date` (initiation date surfaced as `publishedDate`) so newly-reported older-initiated recalls are never skipped; openFDA food/drug/device are **independently cursored** so one failing sub-dataset can't advance over another; a bounded, configurable **recheck window** (`recheckWindowDays`) re-surfaces upstream edits to already-committed records without regressing the cursor; deduplication is now deterministic (order-independent) on equal `publishedDate`; dataset schema hardened + structurally tested. Filters now apply **before** per-source caps; deduplication runs before filtering with a documented merge policy. openFDA full-mode lookback + recency sort and independent food/drug/device sub-source failures; EU full lookback with oldest-first contiguous-week processing, partial-failure reporting and HTTPS host allow-listing + redirect validation; CPSC completeness guard; Health Canada `Organization` exposed as `recallingFirm`; openFDA `countries` no longer contaminated with US state codes. HTTP body-size/timeout limits, HTTP-date `Retry-After`, capped backoff, JSON-parse retries. Strict calendar-date validation, tighter input/dataset schemas, reproducible `npm ci` Docker build. `schemaVersion` → **1.1.0**. - **1.0.0** — Initial release: EU Safety Gate + openFDA + US CPSC + Health Canada; normalized schema, deterministic IDs, dedup, severity normalization, balanced multi-source selection, incremental/full modes, and named-store change tracking. # Actor input Schema ## `sources` (type: `array`): Which official recall sources to query. Defaults to all four (EU + US x2 + Canada). ## `mode` (type: `string`): incremental: only records newer than the last run (or the lookback window on first run). full: a broad backfill over the lookback window. ## `maxItems` (type: `integer`): Upper bound on the number of normalized records saved to the dataset (after filtering and deduplication). ## `lookbackDays` (type: `integer`): How far back to look. Used as the first-run incremental window and as the full-mode backfill window. ## `recheckWindowDays` (type: `integer`): Incremental overlap: re-fetch and re-compare already-committed records whose cursor date is within this many days below the cursor, so upstream edits are surfaced as changed records. The cursor is never regressed by these re-checks. Set 0 to disable. ## `fdaDatasets` (type: `array`): Which openFDA enforcement datasets to include (only relevant when openFDA is selected). ## `product` (type: `string`): Case-insensitive substring match against product name / title / description. ## `brand` (type: `string`): Case-insensitive substring match against brand or recalling firm. ## `category` (type: `string`): Case-insensitive substring match against normalized category labels. ## `country` (type: `string`): Case-insensitive substring match against countries or region scope (e.g. Germany, United States, EU, CA). ## `riskType` (type: `string`): Case-insensitive substring match against the risk type or hazard text (e.g. choking, electric shock, Listeria). ## `severity` (type: `array`): Keep only records whose normalized severity is one of the selected levels. ## `dateFrom` (type: `string`): ISO date, e.g. 2025-01-01. Keeps only records published on/after this date. Invalid calendar dates are rejected. ## `dateTo` (type: `string`): ISO date, e.g. 2025-12-31. Keeps only records published on/before this date. Invalid calendar dates are rejected. ## `since` (type: `string`): Optional ISO date/timestamp overriding the automatic incremental cursor for this run. ## `includeRaw` (type: `boolean`): Include a compact `raw` object of selected upstream values for auditing. Disable for leaner output. ## Actor input object example ```json { "sources": [ "eu_safety_gate", "openfda", "us_cpsc", "health_canada" ], "mode": "incremental", "maxItems": 200, "lookbackDays": 90, "recheckWindowDays": 7, "fdaDatasets": [ "food", "drug", "device" ], "includeRaw": true } ``` # API You can run this Actor programmatically using our API. Below are code examples in JavaScript, Python, and CLI, as well as the OpenAPI specification and MCP server setup. ## JavaScript example ```javascript import { ApifyClient } from 'apify-client'; // Initialize the ApifyClient with your Apify API token // Replace the '' with your token const client = new ApifyClient({ token: '', }); // Prepare Actor input const input = { "sources": [ "eu_safety_gate", "openfda", "us_cpsc", "health_canada" ] }; // Run the Actor and wait for it to finish const run = await client.actor("atlas_digital_assistant/global-product-risk-monitor").call(input); // Fetch and print Actor results from the run's dataset (if any) console.log('Results from dataset'); console.log(`💾 Check your data here: https://console.apify.com/storage/datasets/${run.defaultDatasetId}`); const { items } = await client.dataset(run.defaultDatasetId).listItems(); items.forEach((item) => { console.dir(item); }); // 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/js/docs ``` ## Python example ```python from apify_client import ApifyClient # Initialize the ApifyClient with your Apify API token # Replace '' with your token. client = ApifyClient("") # Prepare the Actor input run_input = { "sources": [ "eu_safety_gate", "openfda", "us_cpsc", "health_canada", ] } # Run the Actor and wait for it to finish run = client.actor("atlas_digital_assistant/global-product-risk-monitor").call(run_input=run_input) # Fetch and print Actor results from the run's dataset (if there are any) print("💾 Check your data here: https://console.apify.com/storage/datasets/" + run["defaultDatasetId"]) for item in client.dataset(run["defaultDatasetId"]).iterate_items(): print(item) # 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/python/docs/quick-start ``` ## CLI example ```bash echo '{ "sources": [ "eu_safety_gate", "openfda", "us_cpsc", "health_canada" ] }' | apify call atlas_digital_assistant/global-product-risk-monitor --silent --output-dataset ``` ## MCP server setup ```json { "mcpServers": { "apify": { "command": "npx", "args": [ "mcp-remote", "https://mcp.apify.com/?tools=atlas_digital_assistant/global-product-risk-monitor", "--header", "Authorization: Bearer " ] } } } ``` ## OpenAPI specification ```json { "openapi": "3.0.1", "info": { "title": "Global Product Risk & Recall Monitor", "description": "Cross-source product-safety recall intelligence: EU Safety Gate, US openFDA, US CPSC and Health Canada normalized into one deduplicated schema with severity, provenance and change tracking.", "version": "1.1", "x-build-id": "h8dBunj4FUlufka9z" }, "servers": [ { "url": "https://api.apify.com/v2" } ], "paths": { "/acts/atlas_digital_assistant~global-product-risk-monitor/run-sync-get-dataset-items": { "post": { "operationId": "run-sync-get-dataset-items-atlas_digital_assistant-global-product-risk-monitor", "x-openai-isConsequential": false, "summary": "Executes an Actor, waits for its completion, and returns Actor's dataset items in response.", "tags": [ "Run Actor" ], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/inputSchema" } } } }, "parameters": [ { "name": "token", "in": "query", "required": true, "schema": { "type": "string" }, "description": "Enter your Apify token here" } ], "responses": { "200": { "description": "OK" } } } }, "/acts/atlas_digital_assistant~global-product-risk-monitor/runs": { "post": { "operationId": "runs-sync-atlas_digital_assistant-global-product-risk-monitor", "x-openai-isConsequential": false, "summary": "Executes an Actor and returns information about the initiated run in response.", "tags": [ "Run Actor" ], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/inputSchema" } } } }, "parameters": [ { "name": "token", "in": "query", "required": true, "schema": { "type": "string" }, "description": "Enter your Apify token here" } ], "responses": { "200": { "description": "OK", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/runsResponseSchema" } } } } } } }, "/acts/atlas_digital_assistant~global-product-risk-monitor/run-sync": { "post": { "operationId": "run-sync-atlas_digital_assistant-global-product-risk-monitor", "x-openai-isConsequential": false, "summary": "Executes an Actor, waits for completion, and returns the OUTPUT from Key-value store in response.", "tags": [ "Run Actor" ], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/inputSchema" } } } }, "parameters": [ { "name": "token", "in": "query", "required": true, "schema": { "type": "string" }, "description": "Enter your Apify token here" } ], "responses": { "200": { "description": "OK" } } } } }, "components": { "schemas": { "inputSchema": { "type": "object", "properties": { "sources": { "title": "Sources", "uniqueItems": true, "type": "array", "description": "Which official recall sources to query. Defaults to all four (EU + US x2 + Canada).", "items": { "type": "string", "enum": [ "eu_safety_gate", "openfda", "us_cpsc", "health_canada" ], "enumTitles": [ "EU Safety Gate (RAPEX) — non-food product alerts", "openFDA Enforcement — food/drug/device recalls (US)", "US CPSC — consumer product recalls", "Health Canada — recalls & safety alerts (Canada)" ] }, "default": [ "eu_safety_gate", "openfda", "us_cpsc", "health_canada" ] }, "mode": { "title": "Run mode", "enum": [ "incremental", "full" ], "type": "string", "description": "incremental: only records newer than the last run (or the lookback window on first run). full: a broad backfill over the lookback window.", "default": "incremental" }, "maxItems": { "title": "Maximum items", "minimum": 1, "maximum": 50000, "type": "integer", "description": "Upper bound on the number of normalized records saved to the dataset (after filtering and deduplication).", "default": 200 }, "lookbackDays": { "title": "Lookback window (days)", "minimum": 1, "maximum": 3650, "type": "integer", "description": "How far back to look. Used as the first-run incremental window and as the full-mode backfill window.", "default": 90 }, "recheckWindowDays": { "title": "Recheck window (days)", "minimum": 0, "maximum": 365, "type": "integer", "description": "Incremental overlap: re-fetch and re-compare already-committed records whose cursor date is within this many days below the cursor, so upstream edits are surfaced as changed records. The cursor is never regressed by these re-checks. Set 0 to disable.", "default": 7 }, "fdaDatasets": { "title": "openFDA datasets", "uniqueItems": true, "type": "array", "description": "Which openFDA enforcement datasets to include (only relevant when openFDA is selected).", "items": { "type": "string", "enum": [ "food", "drug", "device" ], "enumTitles": [ "Food enforcement", "Drug enforcement", "Device enforcement" ] }, "default": [ "food", "drug", "device" ] }, "product": { "title": "Product filter", "type": "string", "description": "Case-insensitive substring match against product name / title / description." }, "brand": { "title": "Brand / firm filter", "type": "string", "description": "Case-insensitive substring match against brand or recalling firm." }, "category": { "title": "Category filter", "type": "string", "description": "Case-insensitive substring match against normalized category labels." }, "country": { "title": "Country filter", "type": "string", "description": "Case-insensitive substring match against countries or region scope (e.g. Germany, United States, EU, CA)." }, "riskType": { "title": "Risk / hazard type filter", "type": "string", "description": "Case-insensitive substring match against the risk type or hazard text (e.g. choking, electric shock, Listeria)." }, "severity": { "title": "Severity filter", "uniqueItems": true, "type": "array", "description": "Keep only records whose normalized severity is one of the selected levels.", "items": { "type": "string", "enum": [ "high", "medium", "low", "unknown" ], "enumTitles": [ "High", "Medium", "Low", "Unknown" ] } }, "dateFrom": { "title": "Date from (inclusive)", "pattern": "^\\d{4}-\\d{2}-\\d{2}([T ].*)?$", "type": "string", "description": "ISO date, e.g. 2025-01-01. Keeps only records published on/after this date. Invalid calendar dates are rejected." }, "dateTo": { "title": "Date to (inclusive)", "pattern": "^\\d{4}-\\d{2}-\\d{2}([T ].*)?$", "type": "string", "description": "ISO date, e.g. 2025-12-31. Keeps only records published on/before this date. Invalid calendar dates are rejected." }, "since": { "title": "Incremental since (override)", "pattern": "^\\d{4}-\\d{2}-\\d{2}([T ].*)?$", "type": "string", "description": "Optional ISO date/timestamp overriding the automatic incremental cursor for this run." }, "includeRaw": { "title": "Include raw source fields", "type": "boolean", "description": "Include a compact `raw` object of selected upstream values for auditing. Disable for leaner output.", "default": true } } }, "runsResponseSchema": { "type": "object", "properties": { "data": { "type": "object", "properties": { "id": { "type": "string" }, "actId": { "type": "string" }, "userId": { "type": "string" }, "startedAt": { "type": "string", "format": "date-time", "example": "2025-01-08T00:00:00.000Z" }, "finishedAt": { "type": "string", "format": "date-time", "example": "2025-01-08T00:00:00.000Z" }, "status": { "type": "string", "example": "READY" }, "meta": { "type": "object", "properties": { "origin": { "type": "string", "example": "API" }, "userAgent": { "type": "string" } } }, "stats": { "type": "object", "properties": { "inputBodyLen": { "type": "integer", "example": 2000 }, "rebootCount": { "type": "integer", "example": 0 }, "restartCount": { "type": "integer", "example": 0 }, "resurrectCount": { "type": "integer", "example": 0 }, "computeUnits": { "type": "integer", "example": 0 } } }, "options": { "type": "object", "properties": { "build": { "type": "string", "example": "latest" }, "timeoutSecs": { "type": "integer", "example": 300 }, "memoryMbytes": { "type": "integer", "example": 1024 }, "diskMbytes": { "type": "integer", "example": 2048 } } }, "buildId": { "type": "string" }, "defaultKeyValueStoreId": { "type": "string" }, "defaultDatasetId": { "type": "string" }, "defaultRequestQueueId": { "type": "string" }, "buildNumber": { "type": "string", "example": "1.0.0" }, "containerUrl": { "type": "string" }, "usage": { "type": "object", "properties": { "ACTOR_COMPUTE_UNITS": { "type": "integer", "example": 0 }, "DATASET_READS": { "type": "integer", "example": 0 }, "DATASET_WRITES": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_READS": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_WRITES": { "type": "integer", "example": 1 }, "KEY_VALUE_STORE_LISTS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_READS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_WRITES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_INTERNAL_GBYTES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_EXTERNAL_GBYTES": { "type": "integer", "example": 0 }, "PROXY_RESIDENTIAL_TRANSFER_GBYTES": { "type": "integer", "example": 0 }, "PROXY_SERPS": { "type": "integer", "example": 0 } } }, "usageTotalUsd": { "type": "number", "example": 0.00005 }, "usageUsd": { "type": "object", "properties": { "ACTOR_COMPUTE_UNITS": { "type": "integer", "example": 0 }, "DATASET_READS": { "type": "integer", "example": 0 }, "DATASET_WRITES": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_READS": { "type": "integer", "example": 0 }, "KEY_VALUE_STORE_WRITES": { "type": "number", "example": 0.00005 }, "KEY_VALUE_STORE_LISTS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_READS": { "type": "integer", "example": 0 }, "REQUEST_QUEUE_WRITES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_INTERNAL_GBYTES": { "type": "integer", "example": 0 }, "DATA_TRANSFER_EXTERNAL_GBYTES": { "type": "integer", "example": 0 }, "PROXY_RESIDENTIAL_TRANSFER_GBYTES": { "type": "integer", "example": 0 }, "PROXY_SERPS": { "type": "integer", "example": 0 } } } } } } } } } } ```