Webhooks (Developer Guide)

const crypto = require('crypto');

function verifyWebhookSignature(payload, signature, secret) {
const expectedSignature = crypto
.createHmac('sha256', secret)
.update(JSON.stringify(payload))
.digest('hex');

return signature === expectedSignature;
}

// In your webhook handler
function handleWebhook(req, res) {
const signature = req.headers['x-kakiyo-signature'];
const payload = req.body;
const secret = 'your_webhook_secret';

if (!verifyWebhookSignature(payload, signature, secret)) {
return res.status(401).send('Invalid signature');
}

// Process the webhook
console.log(`Received event: ${payload.event}`);

// Your webhook handling logic here

res.status(200).send('Webhook received');
}

import hmac
import hashlib
import json

def verify_webhook_signature(payload, signature, secret):
expected_signature = hmac.new(
secret.encode('utf-8'),
json.dumps(payload).encode('utf-8'),
hashlib.sha256
).hexdigest()

return signature == expected_signature

# In your webhook handler (using Flask as an example)
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/kakiyo/webhook', methods=['POST'])
def handle_webhook():
signature = request.headers.get('X-Kakiyo-Signature')
payload = request.json
secret = 'your_webhook_secret'

if not verify_webhook_signature(payload, signature, secret):
return jsonify({'error': 'Invalid signature'}), 401

# Process the webhook
print(f"Received event: {payload['event']}")

# Your webhook handling logic here

return jsonify({'status': 'success'}), 200

const axios = require('axios');

const API_KEY = 'YOUR_API_KEY';
const BASE_URL = 'https://api.kakiyo.com/v1';

async function testWebhook() {
try {
const response = await axios.post(
`${BASE_URL}/webhooks/test`,
{
event: "linkedin.message.received",
data: {
// Optional custom test data
message: "This is a test message"
}
},
{
headers: {
'Authorization': `Bearer ${API_KEY}`,
'Content-Type': 'application/json'
}
}
);

console.log('Test webhook sent:', response.data);
return response.data;
} catch (error) {
console.error('Error testing webhook:', error.response ? error.response.data : error.message);
throw error;
}
}

testWebhook();

import requests

API_KEY = 'YOUR_API_KEY'
BASE_URL = 'https://api.kakiyo.com/v1'

def test_webhook():
try:
response = requests.post(
f'{BASE_URL}/webhooks/test',
json={
'event': 'linkedin.message.received',
'data': {
# Optional custom test data
'message': 'This is a test message'
}
},
headers={
'Authorization': f'Bearer {API_KEY}',
'Content-Type': 'application/json'
}
)

response.raise_for_status()
print('Test webhook sent:', response.json())
return response.json()
except requests.exceptions.RequestException as e:
print('Error testing webhook:', e)
raise

test_webhook()

curl -X POST "https://api.kakiyo.com/v1/webhooks/test" \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"event": "linkedin.message.received",
"data": {
"message": "This is a test message"
}
}'

async function getWebhookExample(eventType) {
try {
const response = await axios.get(
`${BASE_URL}/webhooks/test/example/${eventType}`,
{
headers: {
'Authorization': `Bearer ${API_KEY}`
}
}
);

console.log('Example payload:', response.data);
return response.data;
} catch (error) {
console.error('Error getting example:', error.response ? error.response.data : error.message);
throw error;
}
}

getWebhookExample('linkedin.message.received');

def get_webhook_example(event_type):
try:
response = requests.get(
f'{BASE_URL}/webhooks/test/example/{event_type}',
headers={
'Authorization': f'Bearer {API_KEY}'
}
)

response.raise_for_status()
print('Example payload:', response.json())
return response.json()
except requests.exceptions.RequestException as e:
print('Error getting example:', e)
raise

get_webhook_example('linkedin.message.received')

curl -X GET "https://api.kakiyo.com/v1/webhooks/test/example/linkedin.message.received" \
-H "Authorization: Bearer YOUR_API_KEY"