From palantir-pack
Implements Palantir Foundry webhook handling for ontology changes, dataset updates, and build events using Python/Flask. Covers registration, signature verification, and event routing.
npx claudepluginhub jeremylongshore/claude-code-plugins-plus-skills --plugin palantir-packThis skill is limited to using the following tools:
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.
Generates minimal Python example for Palantir Foundry: query Ontology objects, read datasets, apply actions via SDK. For setup testing or learning API patterns.
Configures Langfuse webhooks for prompt change notifications (created/updated/deleted). Triggers CI/CD on updates or integrates events with Slack/external systems.
Guides Next.js Cache Components and Partial Prerendering (PPR): 'use cache' directives, cacheLife(), cacheTag(), revalidateTag() for caching, invalidation, static/dynamic optimization. Auto-activates on cacheComponents: true.
Share bugs, ideas, or general feedback.
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.
foundry-platform-sdk installedimport 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}")
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
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}")
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
| Issue | Cause | Solution |
|---|---|---|
| Invalid signature | Wrong webhook secret | Verify secret matches registration |
| Timestamp rejected | Server clock drift | Sync NTP; widen tolerance |
| Duplicate events | Network retry | Use event ID deduplication |
| Handler timeout | Slow processing | Offload to background queue |
For performance optimization, see palantir-performance-tuning.