agribackup 1.1.2

AgriBackup Enterprise EUDR API

AgriBackup Python Library

The AgriBackup Python library provides convenient access to the AgriBackup Enterprise API from applications written in Python. It handles cryptographic authentication, deterministic network routing, and executes the complete TRACES NT compliance lifecycle.

Documentation

See the AgriBackup Documentation for Python.

Installation

Install the package via pip:

pip install agribackup

Usage

The package needs to be configured with your account's API key, which you can get at https://www.agribackup.com. The SDK automatically routes your requests to the correct environment (Sandbox or Production) based on your key's prefix.

With this decoupled architecture, the SDK cleanly separates the environment setup (Webhooks) from the actual physical compliance flow.

Phase 1: Pre-assessment (The Sandbox Check)

Action: client.risk_management.assess_coordinate_risk(...)

Purpose: A rapid, synchronous Boolean check to verify if a coordinate is in a deforested zone before you spend capital or compute on heavy satellite ingestion.

from agribackup.client import AgriBackupClient
from agribackup.models import CoordinateRiskRequest
from agribackup.exceptions import ApiException

def assess_coordinate_risk():
    client = AgriBackupClient(api_key="sk_test_YOUR_API_KEY")

    risk_check = client.risk_management.assess_coordinate_risk(
        coordinate_risk_request=CoordinateRiskRequest(latitude=-1.246807, longitude=36.743217)
    )
    
    if risk_check.deforestation_detected:
        print("Deforestation detected. Cannot proceed.")
    else:
        print(f"Coordinate is safe. Risk level: {risk_check.country_risk_level}")
        
    print(risk_check)

if __name__ == "__main__":
    assess_coordinate_risk()

Phase 2: Event-Driven Infrastructure (One-Time Setup)

Action: client.webhooks.register_webhook(...)

Purpose: Establishes the enterprise routing for asynchronous fulfillment. You register your ERP endpoint to listen for polygon.verified, batch.risk_assessed, shipment.linked, and dds.submitted.

# Register your webhook endpoint once during system startup
from agribackup.models import WebhookRegistrationRequest

request = WebhookRegistrationRequest(
    target_url="https://your-erp.internal.co/api/webhooks/agribackup",
    event_types=["batch.risk_assessed", "polygon.verified", "shipment.linked", "dds.generated", "dds.submitted"]
)
registration = client.webhooks.register_webhook(request)
print(f"Webhook Secret (Save securely!): {registration.signing_secret}")

Webhook Event Payloads

Every webhook shares a common envelope (eventType, eventId, timestamp, attempt, nextRetry, data). Below are the schemas for the inner data object for each event:

• polygon.verified: { jobId, polygonsVerified, polygonsFailed, status, polygonIds }
• batch.risk_assessed: { batchId, batchCode, workflowId, riskScore, classification, status }
• shipment.linked: { batchId, shipmentReference, transactionHash }
• dds.generated: { jobId, batchId, ddsReference, status, error }
• dds.submitted: { batchId, ddsReference, status }
• dds.validated: { batchId, ddsReference, validationTimestamp, status }
• dds.rejected: { batchId, ddsReference, rejectionReason, status }
• job.failed: { jobId, jobType, errorCode, errorMessage }
• report.ready: { reportId, reportType, downloadUrl }

Verifying Incoming Webhooks Use your signing_secret to cryptographically verify that incoming webhooks originated from AgriBackup:

import hmac
import hashlib

def verify_webhook(signature_header: str, raw_body_string: str, secret: str) -> bool:
    expected_hash = hmac.new(
        secret.encode('utf-8'),
        raw_body_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected_hash, signature_header)

Alternative: Manual Polling If you prefer not to use webhooks, or need to manually verify a job's status, you can retrieve the job state at any time:

job_response = client.jobs.get_job_status("job-1234-uuid")
phase = job_response.phase

if phase == "COMPLETED":
    print(f"Job completed! Compliant units: {job_response.compliant_units}")
elif phase == "FAILED":
    print(f"Job failed. Errors: {job_response.errors}")
else:
    print(f"Job is still processing. Current phase: {phase}")

Phase 3: The Complete EUDR Execution Lifecycle

This is the core operational loop where the decoupling shines.

1. Ingest Polygons: Call client.polygons.ingest_polygons(...). (Async: wait for polygon.verified webhook).
2. Register Batch: Call client.batches.register_batch(...) using the verified polygon IDs. (Sync: returns batchId instantly).
3. Attach Documentation: Call client.documents.upload(...) or equivalent to bind legal EUDR documents to the batchId.
4. Assess Batch Risk: Call client.batches.assess_risk(...). (Async: wait for batch.risk_assessed webhook).
5. Link Logistics: Call client.logistics.link_batch_to_shipment(...). (Async: wait for shipment.linked webhook).
6. Generate Declaration: Call client.declarations.generate_dds(...). (Async: wait for dds.generated webhook).
7. Submit Declaration: Call client.declarations.submit_dds(...). (Async: wait for TRACES NT dds.validated webhook).

This SDK structure gives you absolute deterministic control over the state machine of your agricultural supply chain.

Complete Lifecycle Implementation Example

from fastapi import FastAPI, Request, BackgroundTasks
from agribackup.client import AgriBackupClient
from agribackup.models import (
    CoordinateRiskRequest, PolygonIngestionRequest, BatchRegistrationRequest, 
    ShipmentLinkRequest, DdsGenerationRequest, GeoJsonFeature, GeoJsonGeometry, FeatureProperties
)
import uvicorn

app = FastAPI()
client = AgriBackupClient(api_key="sk_test_YOUR_API_KEY")

# ==========================================
# Webhook Listener (Handling Async Events)
# ==========================================
@app.post("/api/webhooks/agribackup")
async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
    event = await request.json()
    event_type = event.get("eventType")
    data = event.get("data", {})
    
    if event_type == "polygon.verified":
        background_tasks.add_task(handle_polygon_verified, data)
    elif event_type == "batch.risk_assessed":
        background_tasks.add_task(handle_batch_risk_assessed, data)
    elif event_type == "shipment.linked":
        background_tasks.add_task(handle_shipment_linked, data)
    elif event_type == "dds.generated":
        background_tasks.add_task(handle_dds_generated, data)
    elif event_type == "dds.submitted":
        print(f"DDS Successfully Filed! Reference: {data.get('ddsReference')}")
        
    return {"status": "OK"}

# ==========================================
# The State Machine Workflow
# ==========================================

# Step 1: Ingest Polygons (Triggered manually or via ERP)
def start_compliance_flow():
    polygon_job = client.polygons.ingest_polygons(
        polygon_ingestion_request=PolygonIngestionRequest(
            features=[
                GeoJsonFeature(
                    type="Feature",
                    geometry=GeoJsonGeometry(type="Polygon", coordinates=[[[36.8, -1.2], [36.9, -1.2], [36.9, -1.3], [36.8, -1.3], [36.8, -1.2]]]),
                    properties=FeatureProperties(farmer_name="Global Coffee Farmer #1", farmer_id="TEST_FARMER_100", plot_name="Nyeri Hill Farm Block B", area=2.5, unit="HECTARES", commodity="Coffee")
                )
            ]
        )
    )
    print(f"Polygon ingestion started. Job ID: {polygon_job.job_id}")

# Step 2 & 3: Register Batch & Attach Documents (Fired by polygon.verified webhook)
def handle_polygon_verified(data):
    if data.get("status") != "COMPLETED":
        return
        
    batch = client.batches.register_batch(
        batch_registration_request=BatchRegistrationRequest(
            commodity="Coffee", country_code="KEN", hs_code="0901", quantity_kg=1500.0,
            polygon_ids=data.get("polygonIds", []), vendor_ids=["ERP-VEND-991"]
        )
    )
    
    # Step 4: Assess Batch Risk
    client.batches.assess_risk(batch_id=batch.batch_id)
    print(f"Batch risk assessment started for {batch.batch_id}")

# Step 5: Link Logistics (Fired by batch.risk_assessed webhook)
def handle_batch_risk_assessed(data):
    if data.get("status") != "COMPLETED" or data.get("classification") == "HIGH_RISK":
        return
        
    client.logistics.link_batch_to_shipment(
        shipment_link_request=ShipmentLinkRequest(
            batch_id=data.get("batchId"), shipment_id="SHP-123", bill_of_lading="BOL-99281744", vessel_name="Evergreen"
        )
    )
    print(f"Logistics linked for batch {data.get('batchId')}")

# Step 6: Generate Declaration (Fired by shipment.linked webhook)
def handle_shipment_linked(data):
    # Note: shipment.linked webhook firing indicates success. No status check required.
        
    dds = client.declarations.generate_dds(
        dds_generation_request=DdsGenerationRequest(
            batch_id=data.get("batchId"), legal_document_hashes=["e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"]
        )
    )
    print(f"DDS generation started for batch {data.get('batchId')}. Job ID: {dds.job_id}")
    # Awaits 'dds.generated' webhook...

# Step 7: Submit Declaration (Fired by dds.generated webhook)
def handle_dds_generated(data):
    if data.get("status") != "COMPLETED":
        return
        
    dds_ref = data.get("ddsReference")
    client.declarations.submit_dds(reference_id=dds_ref)
    print(f"DDS submitted to TRACES. Awaiting validation for {dds_ref}...")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Cryptographic Evidence

AgriBackup anchors all critical compliance states to the Hedera Hashgraph DLT. You can retrieve immutable, cryptographically verifiable proofs of your compliance events at any time.

evidence = client.evidence.get_batch_ledger_evidence("batch-123")
print(evidence)

Interpreting the Evidence

The returned evidence object contains a chronological history of the entity's lifecycle anchored on-chain. It includes an array of state_proofs, where each proof represents a distinct compliance event (like CREATED or RISK_ASSESSED).

Key fields inside each state proof include:

• hedera_transaction_id: The exact identifier on the Hedera Consensus Service. You can search this ID on any public Hedera explorer (e.g., Hashscan) to independently verify the transaction.
• consensus_timestamp: The decentralized, network-agreed time the event was permanently recorded.
• operation_type: The specific state transition that occurred.
• merkle_proof: The cryptographic data required to perform offline verification, ensuring the state has not been tampered with since anchoring.

Archiving Reports

AgriBackup supports asynchronous generation of bulk compliance archives. This compiles all XML payloads, Hedera state proofs, and legal document hashes into a single cryptographic ZIP artifact.

1. Request the Archive: Call client.archival.trigger_archive_report with a date range.
2. Await the Webhook: Wait for the report.ready webhook.
3. Download: Use the provided URL or SDK method to securely retrieve the artifact.

from datetime import date
from agribackup.models import ArchiveReportRequest

# Step 1: Request Archive
report_req = client.archival.trigger_archive_report(
    archive_report_request=ArchiveReportRequest(
        start_date=date(2026, 1, 1),
        end_date=date(2026, 3, 31)
    )
)
print(f"Report job started: {report_req.report_id}")

# Step 2: Handle Webhook (fired by report.ready)
def handle_report_ready(data):
    if data.get("reportType") != "COMPLIANCE_ARCHIVE":
        return
    print(f"Report is ready to download at: {data.get('downloadUrl')}")
    
    # Optionally fetch it directly using the SDK:
    # zip_buffer = client.archival.download_archive_report(report_id=data.get("reportId"))

Advanced Enterprise Configuration

Overriding Network Routing & Proxies

The client can be initialized with several options to bypass default network behaviors. This is primarily used by enterprise architectures operating behind zero-trust firewalls or corporate VPC proxies.

client = AgriBackupClient(
    api_key="sk_live_...",
    base_url="https://custom-proxy.internal.co" # Overrides automated prefix routing
)

Manual Idempotency Control

AgriBackup strictly guarantees safety during distributed failures via Idempotency-Key tracking. The backend will automatically generate this key if absent, so standard integrations can safely ignore this parameter.

If your Tier-1 enterprise architecture strictly requires passing your own internal ERP database transaction IDs as idempotency keys, you can inject them securely using the client's default HTTP headers:

# Set the Idempotency-Key globally for the transaction
client.api_client.default_headers["Idempotency-Key"] = "erp-tx-10928-abc"

polygon_job = client.polygons.ingest_polygons(
    polygon_ingestion_request=PolygonIngestionRequest(features=[...])
)

Support

If you require integration support or dedicated VPC configurations, reach out to [e-mail] or visit https://docs.agribackup.com.