--- name: Certyo + Odoo Integration version: 1.0.0 description: Generate Odoo 16+ (Community and Enterprise) to Certyo integration code with field mappings, auth, and working examples api_base: https://www.certyos.com auth: X-API-Key header last_updated: 2026-04-14 --- # Certyo + Odoo Integration Skill This skill enables AI agents to generate production-ready integration code that sends product authenticity, inventory, and manufacturing records from Odoo 16+ (Community and Enterprise editions) to the Certyo blockchain anchoring platform. It covers Python XML-RPC, JSON-RPC, the odoo-await Node.js library, and Automated Action server code approaches. ## Certyo API Reference ### Endpoints | Method | Path | Description | |--------|------|-------------| | `POST` | `/api/v1/records` | Ingest a single record (returns 202 Accepted) | | `POST` | `/api/v1/records/bulk` | Ingest up to 1,000 records in one call | | `POST` | `/api/v1/verify/record` | Verify a record's blockchain anchoring status | | `GET` | `/api/v1/records` | Query records by tenant, database, collection, or recordId | ### Authentication All Certyo API calls require an `X-API-Key` header. ### Record Payload Schema ```json { "tenantId": "string (required)", "database": "string (required)", "collection": "string (required)", "recordId": "string (required)", "recordPayload": { "...any JSON (required)" }, "clientId": "string (optional)", "recordVersion": "string (optional, default '1')", "operationType": "upsert|insert|update|delete (optional)", "sourceTimestamp": "ISO 8601 (optional)", "idempotencyKey": "string (optional)" } ``` ### Successful Response (202 Accepted) ```json { "recordId": "string", "recordHash": "string (SHA-256)", "tenantId": "string", "acceptedAt": "ISO 8601", "idempotencyReplayed": false } ``` ### Pipeline Records flow through: Kafka accumulation (1,000 records or 60 seconds) then Merkle tree computation then IPFS pinning then Polygon blockchain anchor. Typical end-to-end anchoring latency is 60-90 seconds. ## Odoo Integration Pattern ### Architecture ``` Odoo 16+ | | stock.picking state=done / mrp.production state=done / quality.check v Option A: External Middleware (recommended) Python/Node.js service polling ir.cron or listening to webhooks | v Certyo API (POST /api/v1/records) | | (poll for anchored status) v External Middleware --> execute_kw write back to Odoo custom fields Option B: Automated Action (simple deployments) Server Action on stock.picking write(state=done) | | Python code block with urllib/requests v Certyo API (POST /api/v1/records) ``` ### When to Use This Pattern - You are running Odoo 16, 17, or 18 (Community or Enterprise). - You need to anchor delivery orders (stock.picking), manufacturing orders (mrp.production), or quality checks (quality.check). - **Recommended**: Use external middleware (a Python or Node.js service) that polls Odoo via XML-RPC or JSON-RPC and forwards to Certyo. This keeps Odoo clean and avoids HTTP call reliability issues from within Automated Actions. - **Simple deployments**: Use an Automated Action with server code for quick setups, but be aware of timeout and error handling limitations. ### Key Odoo Models | Model | Description | Trigger Condition | |---|---|---| | `stock.picking` | Delivery orders, receipts, internal transfers | `state` changes to `done` | | `mrp.production` | Manufacturing orders | `state` changes to `done` | | `quality.check` | Quality inspection results | `quality_state` changes to `pass` or `fail` | | `stock.lot` | Lot/serial number master | On creation or update | | `stock.move.line` | Detailed stock movements | Created when picking is validated | ## Authentication ### Odoo Side Odoo supports two authentication methods for external API access: **API Keys (Odoo 14+, recommended)** 1. Go to Settings > Users & Companies > Users. 2. Select the integration user. 3. Go to the "API Keys" tab and generate a new key. 4. Use the API key as the password in XML-RPC/JSON-RPC calls. ```python # XML-RPC authentication with API key import xmlrpc.client ODOO_URL = "https://mycompany.odoo.com" ODOO_DB = "mycompany-production" ODOO_USER = "certyo-integration@mycompany.com" ODOO_API_KEY = "" # Not the user password common = xmlrpc.client.ServerProxy(f"{ODOO_URL}/xmlrpc/2/common") uid = common.authenticate(ODOO_DB, ODOO_USER, ODOO_API_KEY, {}) ``` **Session-based (for Automated Actions within Odoo)** Server code in Automated Actions runs as the current user's session. No explicit authentication is needed for Odoo operations; only the Certyo API key is required. ### Certyo Side Store the Certyo API key as an Odoo system parameter: ```python # Store via Settings > Technical > Parameters > System Parameters # Key: certyo.api_key # Value: # Access in server code or external script api_key = env['ir.config_parameter'].sudo().get_param('certyo.api_key') ``` ## Field Mapping ### stock.picking (Delivery Orders) | Odoo Field | Certyo Field | Notes | |---|---|---| | `stock.picking.name` | `recordId` | Transfer reference (e.g., "WH/OUT/00042") | | `stock.picking.date_done` | `sourceTimestamp` | Completion timestamp, convert to ISO 8601 | | `"odoo"` (constant) | `database` | Identifies the source system | | `"stock.picking"` (constant) | `collection` | Identifies the document type | | `stock.move.line.product_id.default_code` | `recordPayload.sku` | Product internal reference | | `stock.move.line.product_id.name` | `recordPayload.productName` | Product display name | | `stock.move.line.quantity` (Odoo 17+) or `stock.move.line.qty_done` (Odoo 16) | `recordPayload.quantity` | Done quantity (field name varies by version) | | `stock.move.line.lot_id.name` | `recordPayload.lotNumber` | Lot/serial number | | `stock.picking.partner_id.name` | `recordPayload.partnerName` | Customer or supplier | | `stock.picking.origin` | `recordPayload.sourceDocument` | Source sales order reference | | `stock.picking.picking_type_id.code` | `recordPayload.operationType` | incoming, outgoing, internal | | Odoo database name | `tenantId` | Your Certyo tenant identifier | ### mrp.production (Manufacturing Orders) | Odoo Field | Certyo Field | Notes | |---|---|---| | `mrp.production.name` | `recordId` | Production order reference (e.g., "MO/00015") | | `mrp.production.date_finished` | `sourceTimestamp` | Completion timestamp | | `"odoo"` (constant) | `database` | | | `"mrp.production"` (constant) | `collection` | | | `mrp.production.product_id.default_code` | `recordPayload.sku` | Finished product SKU | | `mrp.production.qty_produced` (Odoo 17+) or `mrp.production.qty_producing` (Odoo 16) | `recordPayload.quantityProduced` | Version-dependent field | | `mrp.production.lot_producing_id.name` | `recordPayload.lotNumber` | Lot assigned to finished product | ## Code Examples ### Example 1: Python XML-RPC External Middleware A complete external service that polls Odoo for completed deliveries and sends them to Certyo. ```python """ Odoo to Certyo XML-RPC Middleware Polls for completed stock.picking records and sends them to Certyo. Runs as an external service (systemd, Docker, etc.) """ import os import time import logging from datetime import datetime, timezone import xmlrpc.client import requests logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger(__name__) # Odoo connection ODOO_URL = os.environ["ODOO_URL"] ODOO_DB = os.environ["ODOO_DB"] ODOO_USER = os.environ["ODOO_USER"] ODOO_API_KEY = os.environ["ODOO_API_KEY"] # Certyo connection CERTYO_BASE = "https://www.certyos.com" CERTYO_API_KEY = os.environ["CERTYO_API_KEY"] CERTYO_TENANT_ID = os.environ["CERTYO_TENANT_ID"] # Polling interval POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", "60")) # Odoo version detection (affects field names) ODOO_VERSION = int(os.environ.get("ODOO_VERSION", "17")) def connect_odoo(): common = xmlrpc.client.ServerProxy(f"{ODOO_URL}/xmlrpc/2/common") uid = common.authenticate(ODOO_DB, ODOO_USER, ODOO_API_KEY, {}) if not uid: raise ConnectionError("Odoo authentication failed.") models = xmlrpc.client.ServerProxy(f"{ODOO_URL}/xmlrpc/2/object") return uid, models def get_qty_done_field() -> str: """Return the correct quantity field name based on Odoo version.""" if ODOO_VERSION >= 17: return "quantity" return "qty_done" def poll_completed_pickings(uid: int, models, last_check: str) -> list: """Find stock.picking records that are done and not yet sent to Certyo.""" domain = [ ("state", "=", "done"), ("x_certyo_status", "in", [False, ""]), ("date_done", ">=", last_check), ] picking_ids = models.execute_kw( ODOO_DB, uid, ODOO_API_KEY, "stock.picking", "search", [domain], {"limit": 500}, ) if not picking_ids: return [] pickings = models.execute_kw( ODOO_DB, uid, ODOO_API_KEY, "stock.picking", "read", [picking_ids], {"fields": [ "name", "date_done", "partner_id", "origin", "picking_type_id", "move_line_ids", ]}, ) return pickings def get_move_lines(uid: int, models, line_ids: list) -> list: """Read move line details including product, lot, and quantity.""" qty_field = get_qty_done_field() lines = models.execute_kw( ODOO_DB, uid, ODOO_API_KEY, "stock.move.line", "read", [line_ids], {"fields": ["product_id", "lot_id", qty_field]}, ) result = [] for line in lines: product_id = line["product_id"][0] if line["product_id"] else False product_data = {} if product_id: products = models.execute_kw( ODOO_DB, uid, ODOO_API_KEY, "product.product", "read", [[product_id]], {"fields": ["default_code", "name"]}, ) if products: product_data = products[0] result.append({ "sku": product_data.get("default_code", ""), "productName": product_data.get("name", ""), "quantity": line.get(qty_field, 0), "lotNumber": line["lot_id"][1] if line.get("lot_id") else "", }) return result def transform_picking(uid: int, models, picking: dict) -> dict: """Transform an Odoo picking into a Certyo record.""" move_lines = get_move_lines(uid, models, picking.get("move_line_ids", [])) date_done = picking.get("date_done", "") if isinstance(date_done, str) and date_done: source_timestamp = date_done.replace(" ", "T") + "Z" else: source_timestamp = datetime.now(timezone.utc).isoformat() picking_type_code = "" if picking.get("picking_type_id"): pt = models.execute_kw( ODOO_DB, uid, ODOO_API_KEY, "stock.picking.type", "read", [[picking["picking_type_id"][0]]], {"fields": ["code"]}, ) if pt: picking_type_code = pt[0].get("code", "") return { "tenantId": CERTYO_TENANT_ID, "database": "odoo", "collection": "stock.picking", "recordId": picking["name"], "operationType": "insert", "sourceTimestamp": source_timestamp, "idempotencyKey": f"{picking['name']}-v1", "recordPayload": { "pickingName": picking["name"], "partnerName": picking["partner_id"][1] if picking.get("partner_id") else "", "sourceDocument": picking.get("origin", ""), "pickingTypeCode": picking_type_code, "lines": move_lines, }, } def send_to_certyo(records: list) -> list: """Send records to Certyo. Returns list of (recordId, recordHash) tuples.""" if not records: return [] headers = {"Content-Type": "application/json", "X-API-Key": CERTYO_API_KEY} results = [] if len(records) == 1: resp = requests.post( f"{CERTYO_BASE}/api/v1/records", json=records[0], headers=headers, timeout=10, ) resp.raise_for_status() body = resp.json() results.append((body["recordId"], body["recordHash"])) else: for i in range(0, len(records), 1000): batch = records[i: i + 1000] resp = requests.post( f"{CERTYO_BASE}/api/v1/records/bulk", json=batch, headers=headers, timeout=30, ) resp.raise_for_status() # Bulk returns list of results for item in resp.json(): results.append((item["recordId"], item["recordHash"])) return results def write_back_to_odoo(uid: int, models, picking_name: str, record_hash: str) -> None: """Write Certyo hash and status back to the Odoo picking.""" picking_ids = models.execute_kw( ODOO_DB, uid, ODOO_API_KEY, "stock.picking", "search", [[("name", "=", picking_name)]], ) if picking_ids: models.execute_kw( ODOO_DB, uid, ODOO_API_KEY, "stock.picking", "write", [picking_ids, { "x_certyo_record_hash": record_hash, "x_certyo_anchor_status": "Ingested", }], ) logger.info("Wrote Certyo hash back to Odoo picking %s.", picking_name) def main(): uid, models = connect_odoo() last_check = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") logger.info("Starting Odoo-Certyo middleware (version: %d, interval: %ds)", ODOO_VERSION, POLL_INTERVAL) while True: try: pickings = poll_completed_pickings(uid, models, last_check) logger.info("Found %d completed pickings since %s.", len(pickings), last_check) if pickings: records = [transform_picking(uid, models, p) for p in pickings] results = send_to_certyo(records) for record_id, record_hash in results: write_back_to_odoo(uid, models, record_id, record_hash) last_check = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") except Exception: logger.exception("Poll cycle failed. Will retry next interval.") time.sleep(POLL_INTERVAL) if __name__ == "__main__": main() ``` ### Example 2: Python JSON-RPC Approach An alternative using Odoo's JSON-RPC endpoint, which is the native protocol for the Odoo web client. ```python """ Odoo JSON-RPC client for Certyo integration. JSON-RPC is Odoo's native protocol and can be more reliable than XML-RPC for Odoo.sh and cloud-hosted instances. """ import os import json import logging import requests logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) ODOO_URL = os.environ["ODOO_URL"] ODOO_DB = os.environ["ODOO_DB"] ODOO_USER = os.environ["ODOO_USER"] ODOO_API_KEY = os.environ["ODOO_API_KEY"] CERTYO_BASE = "https://www.certyos.com" CERTYO_API_KEY = os.environ["CERTYO_API_KEY"] CERTYO_TENANT_ID = os.environ["CERTYO_TENANT_ID"] ODOO_VERSION = int(os.environ.get("ODOO_VERSION", "17")) _request_id = 0 def jsonrpc_call(url: str, method: str, params: dict) -> dict: """Make a JSON-RPC call to Odoo.""" global _request_id _request_id += 1 payload = { "jsonrpc": "2.0", "method": method, "id": _request_id, "params": params, } resp = requests.post(url, json=payload, timeout=30) resp.raise_for_status() result = resp.json() if result.get("error"): raise Exception(f"JSON-RPC error: {result['error']}") return result.get("result") def authenticate() -> int: """Authenticate and return uid.""" result = jsonrpc_call( f"{ODOO_URL}/jsonrpc", "call", { "service": "common", "method": "authenticate", "args": [ODOO_DB, ODOO_USER, ODOO_API_KEY, {}], }, ) return result def search_read(uid: int, model: str, domain: list, fields: list, limit: int = 100) -> list: """Search and read records from Odoo.""" return jsonrpc_call( f"{ODOO_URL}/jsonrpc", "call", { "service": "object", "method": "execute_kw", "args": [ ODOO_DB, uid, ODOO_API_KEY, model, "search_read", [domain], {"fields": fields, "limit": limit}, ], }, ) def write_record(uid: int, model: str, record_ids: list, values: dict) -> bool: """Write values to Odoo records.""" return jsonrpc_call( f"{ODOO_URL}/jsonrpc", "call", { "service": "object", "method": "execute_kw", "args": [ ODOO_DB, uid, ODOO_API_KEY, model, "write", [record_ids, values], ], }, ) def sync_pickings(): """Sync completed pickings to Certyo using JSON-RPC.""" uid = authenticate() logger.info("Authenticated as uid %d.", uid) qty_field = "quantity" if ODOO_VERSION >= 17 else "qty_done" pickings = search_read( uid, "stock.picking", [("state", "=", "done"), ("x_certyo_anchor_status", "in", [False, ""])], ["name", "date_done", "partner_id", "origin", "move_line_ids"], limit=200, ) logger.info("Found %d pickings to sync.", len(pickings)) certyo_records = [] for picking in pickings: lines = search_read( uid, "stock.move.line", [("id", "in", picking["move_line_ids"])], ["product_id", "lot_id", qty_field], ) items = [] for line in lines: items.append({ "productId": line["product_id"][0] if line.get("product_id") else None, "productName": line["product_id"][1] if line.get("product_id") else "", "lotNumber": line["lot_id"][1] if line.get("lot_id") else "", "quantity": line.get(qty_field, 0), }) date_done = picking.get("date_done", "") source_ts = f"{date_done}Z" if date_done else "" certyo_records.append({ "tenantId": CERTYO_TENANT_ID, "database": "odoo", "collection": "stock.picking", "recordId": picking["name"], "operationType": "insert", "sourceTimestamp": source_ts.replace(" ", "T"), "idempotencyKey": f"{picking['name']}-v1", "recordPayload": { "pickingName": picking["name"], "partnerName": picking["partner_id"][1] if picking.get("partner_id") else "", "sourceDocument": picking.get("origin", ""), "lines": items, }, }) if not certyo_records: logger.info("No records to send.") return # Send to Certyo headers = {"Content-Type": "application/json", "X-API-Key": CERTYO_API_KEY} if len(certyo_records) <= 1: resp = requests.post( f"{CERTYO_BASE}/api/v1/records", json=certyo_records[0], headers=headers, timeout=10, ) else: resp = requests.post( f"{CERTYO_BASE}/api/v1/records/bulk", json=certyo_records, headers=headers, timeout=30, ) resp.raise_for_status() logger.info("Sent %d records to Certyo. Status: %d", len(certyo_records), resp.status_code) # Write back to Odoo for picking in pickings: write_record( uid, "stock.picking", [picking["id"]], {"x_certyo_anchor_status": "Ingested"}, ) if __name__ == "__main__": sync_pickings() ``` ### Example 3: Node.js with odoo-await Library For teams that prefer JavaScript/TypeScript, using the `odoo-await` npm package. ```javascript /** * Odoo to Certyo sync using odoo-await. * Install: npm install odoo-await axios */ const Odoo = require("odoo-await"); const axios = require("axios"); const CERTYO_BASE = "https://www.certyos.com"; const CERTYO_API_KEY = process.env.CERTYO_API_KEY; const CERTYO_TENANT_ID = process.env.CERTYO_TENANT_ID; const ODOO_VERSION = parseInt(process.env.ODOO_VERSION || "17", 10); async function main() { // Connect to Odoo const odoo = new Odoo({ baseUrl: process.env.ODOO_URL, db: process.env.ODOO_DB, username: process.env.ODOO_USER, password: process.env.ODOO_API_KEY, }); await odoo.connect(); console.log("Connected to Odoo."); // Find completed pickings not yet sent to Certyo const pickings = await odoo.searchRead("stock.picking", { domain: [ ["state", "=", "done"], ["x_certyo_anchor_status", "in", [false, ""]], ], fields: ["name", "date_done", "partner_id", "origin", "move_line_ids"], limit: 200, }); console.log(`Found ${pickings.length} pickings to sync.`); const qtyField = ODOO_VERSION >= 17 ? "quantity" : "qty_done"; const certYoRecords = []; for (const picking of pickings) { const lines = await odoo.read("stock.move.line", picking.move_line_ids, [ "product_id", "lot_id", qtyField, ]); const items = lines.map((line) => ({ productName: line.product_id ? line.product_id[1] : "", lotNumber: line.lot_id ? line.lot_id[1] : "", quantity: line[qtyField] || 0, })); const dateDone = picking.date_done || ""; const sourceTimestamp = dateDone ? dateDone.replace(" ", "T") + "Z" : new Date().toISOString(); certYoRecords.push({ tenantId: CERTYO_TENANT_ID, database: "odoo", collection: "stock.picking", recordId: picking.name, operationType: "insert", sourceTimestamp, idempotencyKey: `${picking.name}-v1`, recordPayload: { pickingName: picking.name, partnerName: picking.partner_id ? picking.partner_id[1] : "", sourceDocument: picking.origin || "", lines: items, }, }); } if (certYoRecords.length === 0) { console.log("No records to send."); return; } // Send to Certyo const endpoint = certYoRecords.length > 1 ? "/api/v1/records/bulk" : "/api/v1/records"; const payload = certYoRecords.length > 1 ? certYoRecords : certYoRecords[0]; const response = await axios.post(`${CERTYO_BASE}${endpoint}`, payload, { headers: { "Content-Type": "application/json", "X-API-Key": CERTYO_API_KEY, }, timeout: 30000, }); console.log( `Sent ${certYoRecords.length} records to Certyo. Status: ${response.status}` ); // Write back to Odoo for (const picking of pickings) { await odoo.update("stock.picking", picking.id, { x_certyo_anchor_status: "Ingested", }); } console.log("Write-back complete."); } main().catch(console.error); ``` ### Example 4: Odoo Automated Action Server Code For simple deployments where external middleware is not available. This runs inside Odoo when a stock.picking transitions to `done`. ```python # Automated Action Configuration: # Model: stock.picking # Trigger: On Update # Before Update Filter: [("state", "!=", "done")] # Filter: [("state", "=", "done")] # Action Type: Execute Python Code # Available variables: env, model, record, records, time, datetime, dateutil, # timezone, float_compare, float_round, float_is_zero, # UserError, Command, log import json from urllib.request import Request, urlopen from urllib.error import URLError, HTTPError CERTYO_BASE = "https://www.certyos.com" for picking in records: # Skip if already sent if picking.x_certyo_anchor_status: continue # Get Certyo credentials from system parameters api_key = env["ir.config_parameter"].sudo().get_param("certyo.api_key") tenant_id = env["ir.config_parameter"].sudo().get_param("certyo.tenant_id") if not api_key or not tenant_id: log("Certyo credentials not configured in system parameters.", level="error") continue # Determine the correct quantity field based on Odoo version # In Odoo 17+, stock.move.line uses 'quantity' instead of 'qty_done' qty_field = "quantity" if hasattr(picking.move_line_ids[:1], "quantity") else "qty_done" # Build line items lines = [] for line in picking.move_line_ids: qty = getattr(line, qty_field, 0) lines.append({ "sku": line.product_id.default_code or "", "productName": line.product_id.name or "", "quantity": qty, "lotNumber": line.lot_id.name if line.lot_id else "", }) # Build Certyo payload date_done = picking.date_done if date_done: source_timestamp = date_done.strftime("%Y-%m-%dT%H:%M:%SZ") else: source_timestamp = datetime.datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ") payload = { "tenantId": tenant_id, "database": "odoo", "collection": "stock.picking", "recordId": picking.name, "operationType": "insert", "sourceTimestamp": source_timestamp, "idempotencyKey": "%s-v1" % picking.name, "recordPayload": { "pickingName": picking.name, "partnerName": picking.partner_id.name if picking.partner_id else "", "sourceDocument": picking.origin or "", "pickingTypeCode": picking.picking_type_id.code or "", "lines": lines, }, } # Send to Certyo data = json.dumps(payload).encode("utf-8") req = Request( CERTYO_BASE + "/api/v1/records", data=data, headers={ "Content-Type": "application/json", "X-API-Key": api_key, }, method="POST", ) try: with urlopen(req, timeout=10) as resp: body = json.loads(resp.read().decode("utf-8")) picking.write({ "x_certyo_record_hash": body.get("recordHash", ""), "x_certyo_anchor_status": "Ingested", }) log("Certyo: ingested %s, hash=%s" % (picking.name, body.get("recordHash", ""))) except (URLError, HTTPError) as e: log("Certyo: failed to ingest %s: %s" % (picking.name, str(e)), level="error") # Do not raise - allow the picking state change to complete ``` ## Verification and Write-back After Certyo anchors records (typically 60-90 seconds), verify and update the Odoo custom fields. ### Custom Fields Required Add the following custom fields to `stock.picking` (and optionally `mrp.production`): | Technical Name | Label | Type | Size | |---|---|---|---| | `x_certyo_record_hash` | Certyo Record Hash | Char | 64 | | `x_certyo_anchor_status` | Certyo Anchor Status | Selection | Ingested, Anchored, Failed | | `x_certyo_anchored_at` | Certyo Anchored At | Datetime | - | Create these via Settings > Technical > Database Structure > Fields, or in a custom module: ```python # In a custom Odoo module's models.py from odoo import fields, models class StockPicking(models.Model): _inherit = "stock.picking" x_certyo_record_hash = fields.Char(string="Certyo Record Hash", size=64, readonly=True) x_certyo_anchor_status = fields.Selection( [("ingested", "Ingested"), ("anchored", "Anchored"), ("failed", "Failed")], string="Certyo Anchor Status", readonly=True, ) x_certyo_anchored_at = fields.Datetime(string="Certyo Anchored At", readonly=True) ``` ### Verification Cron Job Set up an `ir.cron` scheduled action to poll Certyo for anchoring status: ```python # Scheduled Action (ir.cron) - runs every 5 minutes # Model: stock.picking # Action Type: Execute Python Code import json from urllib.request import Request, urlopen CERTYO_BASE = "https://www.certyos.com" api_key = env["ir.config_parameter"].sudo().get_param("certyo.api_key") tenant_id = env["ir.config_parameter"].sudo().get_param("certyo.tenant_id") # Find pickings waiting for anchoring pickings = env["stock.picking"].search([ ("x_certyo_anchor_status", "=", "ingested"), ], limit=50) for picking in pickings: payload = json.dumps({ "tenantId": tenant_id, "database": "odoo", "collection": "stock.picking", "recordId": picking.name, }).encode("utf-8") req = Request( CERTYO_BASE + "/api/v1/verify/record", data=payload, headers={ "Content-Type": "application/json", "X-API-Key": api_key, }, method="POST", ) try: with urlopen(req, timeout=10) as resp: body = json.loads(resp.read().decode("utf-8")) if body.get("verified"): picking.write({ "x_certyo_anchor_status": "anchored", "x_certyo_anchored_at": datetime.datetime.now(), }) log("Certyo: %s verified and anchored." % picking.name) except Exception as e: log("Certyo: verification failed for %s: %s" % (picking.name, str(e)), level="error") ``` ## Code Generation Rules 1. **Prefer external middleware over Automated Actions for HTTP calls.** Automated Actions run inside the Odoo transaction. If the HTTP call to Certyo times out or fails, it can block or slow down the user's workflow. External middleware (Python service, Node.js, etc.) decouples the systems. 2. **Use `ir.cron` for polling patterns.** When using external middleware is not possible, set up an `ir.cron` scheduled action that runs every few minutes to find new completed pickings and send them to Certyo, rather than relying solely on real-time Automated Actions. 3. **Store credentials as system parameters.** Use `ir.config_parameter` (Settings > Technical > System Parameters) with keys like `certyo.api_key` and `certyo.tenant_id`. Access them with `env['ir.config_parameter'].sudo().get_param(...)`. Never hardcode credentials. 4. **Handle Odoo version differences for quantity fields.** In Odoo 17+, `stock.move.line` uses `quantity` instead of `qty_done`. Similarly, `mrp.production` uses `qty_produced` in 17+ versus `qty_producing` in 16. Always detect the version or use `hasattr()` to check field existence. 5. **Use `x_` prefix for custom fields.** Odoo convention for fields added outside a module is the `x_` prefix (e.g., `x_certyo_record_hash`). If creating a proper Odoo module, you can omit the prefix (e.g., `certyo_record_hash`) but must include it in the module's `__manifest__.py`. 6. **Batch with the bulk endpoint when processing more than 10 records.** When the middleware polls and finds many completed pickings, accumulate them and use `POST /api/v1/records/bulk` (up to 1,000 per call) instead of individual calls. 7. **Always set `idempotencyKey`.** Use the pattern `{picking.name}-v1`. Odoo can trigger Automated Actions multiple times if a record is written to in the same transaction. The idempotency key prevents duplicate Certyo records. 8. **Do not raise exceptions in Automated Actions.** If the Certyo API call fails, log the error and continue. Raising an exception will roll back the entire Odoo transaction, preventing the stock.picking from being marked as done. 9. **Use `sudo()` sparingly and only for system parameter access.** When reading `ir.config_parameter`, use `sudo()` because system parameters require admin access. Do not use `sudo()` for reading or writing `stock.picking` records unless the integration user lacks the necessary permissions. 10. **In Automated Actions, use `urllib` instead of `requests`.** The `requests` library may not be available in all Odoo deployments (especially Odoo.sh sandboxed environments). Use Python's built-in `urllib.request` for HTTP calls in server action code. External middleware can freely use `requests`.