Palantir Webhooks Events

Palantir Webhooks & Events

Overview

Handle Foundry webhook events for Ontology changes, dataset updates, and build completions. Covers webhook registration via the Foundry API, signature verification, event routing, and idempotent processing.

Prerequisites

• Foundry enrollment with webhook support enabled
• HTTPS endpoint accessible from Foundry's network
• foundry-platform-sdk installed

Instructions

Step 1: Register a Webhook via API

import os, foundry

client = foundry.FoundryClient(
    auth=foundry.ConfidentialClientAuth(
        client_id=os.environ["FOUNDRY_CLIENT_ID"],
        client_secret=os.environ["FOUNDRY_CLIENT_SECRET"],
        hostname=os.environ["FOUNDRY_HOSTNAME"],
        scopes=["api:read-data", "api:write-data"],
    ),
    hostname=os.environ["FOUNDRY_HOSTNAME"],
)

# Register webhook for object change events
webhook = client.webhooks.Webhook.create(
    url="https://myapp.example.com/webhooks/foundry",
    event_types=["ontology.object.created", "ontology.object.updated"],
    secret="whsec_your_webhook_secret_here",
)
print(f"Webhook registered: {webhook.rid}")

Step 2: Webhook Endpoint with Signature Verification

from flask import Flask, request, jsonify
import hmac, hashlib

app = Flask(__name__)

@app.post("/webhooks/foundry")
def handle_foundry_webhook():
    # Verify signature
    signature = request.headers.get("X-Foundry-Signature", "")
    timestamp = request.headers.get("X-Foundry-Timestamp", "")
    secret = os.environ["FOUNDRY_WEBHOOK_SECRET"]

    signed_payload = f"{timestamp}.{request.get_data(as_text=True)}"
    expected = hmac.new(
        secret.encode(), signed_payload.encode(), hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "Invalid signature"}), 401

    # Replay protection — reject timestamps older than 5 minutes
    import time
    if abs(time.time() - int(timestamp)) > 300:
        return jsonify({"error": "Timestamp too old"}), 401

    event = request.get_json()
    handle_event(event)
    return jsonify({"received": True}), 200

Step 3: Event Router

def handle_event(event: dict):
    event_type = event.get("type", "")
    handlers = {
        "ontology.object.created": on_object_created,
        "ontology.object.updated": on_object_updated,
        "ontology.object.deleted": on_object_deleted,
        "dataset.updated": on_dataset_updated,
        "build.completed": on_build_completed,
    }
    handler = handlers.get(event_type)
    if handler:
        handler(event["data"])
    else:
        print(f"Unhandled event type: {event_type}")

def on_object_created(data: dict):
    obj_type = data["objectType"]
    primary_key = data["primaryKey"]
    print(f"Object created: {obj_type}/{primary_key}")
    # Sync to external system, trigger workflow, etc.

def on_object_updated(data: dict):
    obj_type = data["objectType"]
    changes = data.get("changedProperties", {})
    print(f"Object updated: {obj_type} — changed: {list(changes.keys())}")

def on_object_deleted(data: dict):
    print(f"Object deleted: {data['objectType']}/{data['primaryKey']}")

def on_dataset_updated(data: dict):
    print(f"Dataset updated: {data['datasetRid']} branch={data['branch']}")

def on_build_completed(data: dict):
    status = data["buildStatus"]
    print(f"Build {data['buildRid']}: {status}")

Step 4: Idempotent Processing

import redis

r = redis.Redis.from_url(os.environ.get("REDIS_URL", "redis://localhost:6379"))

def idempotent_handle(event: dict):
    event_id = event["id"]
    key = f"foundry:event:{event_id}"
    if r.exists(key):
        print(f"Skipping duplicate event: {event_id}")
        return
    handle_event(event)
    r.setex(key, 86400 * 7, "processed")  # 7-day TTL

Output

• Webhook registered with Foundry for Ontology/dataset events
• Signature verification with replay protection
• Event router dispatching to typed handlers
• Idempotent processing preventing duplicate handling

Error Handling

Resources

• Foundry API Reference
• Foundry Documentation

Next Steps

For performance optimization, see palantir-performance-tuning.