zhongwei/gh-jeremylongshore-claude-code-plugins-plus-plugins-api-development-webhook-handler-creator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
Zhongwei Li
2025-11-29 18:52:47 +08:00
commit 282d2cbece
17 changed files with 1973 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
skills/skill-adapter/assets/README.md Normal file
View File

@@ -0,0 +1,7 @@
# Assets
Bundled resources for webhook-handler-creator skill
- [ ] webhook_handler_template.py: A basic Python template for a webhook handler with signature verification and idempotency.
- [ ] example_webhook_payload.json: Example JSON payload for a webhook.
- [ ] example_event_schema.json: Example JSON schema for a webhook event.

32
skills/skill-adapter/assets/config-template.json Normal file
View File

@@ -0,0 +1,32 @@
{
"skill": {
"name": "skill-name",
"version": "1.0.0",
"enabled": true,
"settings": {
"verbose": false,
"autoActivate": true,
"toolRestrictions": true
}
},
"triggers": {
"keywords": [
"example-trigger-1",
"example-trigger-2"
],
"patterns": []
},
"tools": {
"allowed": [
"Read",
"Grep",
"Bash"
],
"restricted": []
},
"metadata": {
"author": "Plugin Author",
"category": "general",
"tags": []
}
}

125
skills/skill-adapter/assets/example_event_schema.json Normal file
View File

@@ -0,0 +1,125 @@
{
"_comment": "Example JSON schema for a webhook event related to order processing.",
"type": "object",
"properties": {
"event_type": {
"type": "string",
"description": "Type of event. e.g., order.created, order.updated, order.cancelled",
"example": "order.created"
},
"event_id": {
"type": "string",
"format": "uuid",
"description": "Unique identifier for the event.",
"example": "a1b2c3d4-e5f6-7890-1234-567890abcdef"
},
"timestamp": {
"type": "string",
"format": "date-time",
"description": "Timestamp of when the event occurred (ISO 8601 format).",
"example": "2024-01-27T12:00:00Z"
},
"data": {
"type": "object",
"description": "The actual data associated with the event.",
"properties": {
"order_id": {
"type": "string",
"description": "Unique identifier for the order.",
"example": "ORD-2024-001"
},
"customer_id": {
"type": "string",
"description": "Unique identifier for the customer.",
"example": "CUST-001"
},
"amount": {
"type": "number",
"format": "float",
"description": "Total order amount.",
"example": 99.99
},
"currency": {
"type": "string",
"description": "Currency code (ISO 4217).",
"example": "USD"
},
"status": {
"type": "string",
"description": "Current status of the order.",
"example": "pending",
"enum": ["pending", "processing", "shipped", "delivered", "cancelled"]
},
"shipping_address": {
"type": "object",
"description": "Shipping address details.",
"properties": {
"street": {
"type": "string",
"example": "123 Main St"
},
"city": {
"type": "string",
"example": "Anytown"
},
"state": {
"type": "string",
"example": "CA"
},
"zip": {
"type": "string",
"example": "91234"
},
"country": {
"type": "string",
"example": "USA"
}
},
"required": ["street", "city", "state", "zip", "country"]
},
"items": {
"type": "array",
"description": "List of items in the order.",
"items": {
"type": "object",
"properties": {
"product_id": {
"type": "string",
"example": "PROD-001"
},
"quantity": {
"type": "integer",
"example": 2
},
"price": {
"type": "number",
"format": "float",
"example": 49.99
}
},
"required": ["product_id", "quantity", "price"]
}
}
},
"required": ["order_id", "customer_id", "amount", "currency", "status", "shipping_address", "items"]
},
"metadata": {
"type": "object",
"description": "Additional metadata about the event.",
"properties": {
"source": {
"type": "string",
"description": "Source of the event (e.g., 'payment_gateway', 'crm').",
"example": "payment_gateway"
},
"version": {
"type": "string",
"description": "Version of the event schema.",
"example": "1.0"
}
},
"required": ["source", "version"]
}
},
"required": ["event_type", "event_id", "timestamp", "data", "metadata"]
}

53
skills/skill-adapter/assets/example_webhook_payload.json Normal file
View File

@@ -0,0 +1,53 @@
{
"_comment": "Example JSON payload for a webhook. This payload represents a common event: an order being placed.",
"event_type": "order.created",
"_comment": "Type of event that triggered the webhook.",
"event_id": "evt_1234567890abcdef",
"_comment": "Unique identifier for this specific event. Used for idempotency.",
"timestamp": 1678886400,
"_comment": "Unix timestamp of when the event occurred.",
"data": {
"_comment": "The actual data associated with the event. Structure varies depending on event_type.",
"order_id": "ord_abcdef1234567890",
"customer_id": "cust_fedcba0987654321",
"order_total": 99.99,
"currency": "USD",
"items": [
{
"item_id": "item_11223344",
"quantity": 1,
"price": 49.99,
"name": "Awesome Widget"
},
{
"item_id": "item_55667788",
"quantity": 1,
"price": 50.00,
"name": "Deluxe Gadget"
}
],
"shipping_address": {
"name": "John Doe",
"address_line1": "123 Main St",
"address_line2": null,
"city": "Anytown",
"state": "CA",
"zip_code": "91234",
"country": "US"
},
"billing_address": {
"name": "John Doe",
"address_line1": "123 Main St",
"address_line2": null,
"city": "Anytown",
"state": "CA",
"zip_code": "91234",
"country": "US"
}
},
"metadata": {
"_comment": "Optional metadata associated with the event. Can be used to pass additional context.",
"source": "web",
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36"
}
}

28
skills/skill-adapter/assets/skill-schema.json Normal file
View File

@@ -0,0 +1,28 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "Claude Skill Configuration",
"type": "object",
"required": ["name", "description"],
"properties": {
"name": {
"type": "string",
"pattern": "^[a-z0-9-]+$",
"maxLength": 64,
"description": "Skill identifier (lowercase, hyphens only)"
},
"description": {
"type": "string",
"maxLength": 1024,
"description": "What the skill does and when to use it"
},
"allowed-tools": {
"type": "string",
"description": "Comma-separated list of allowed tools"
},
"version": {
"type": "string",
"pattern": "^\\d+\\.\\d+\\.\\d+$",
"description": "Semantic version (x.y.z)"
}
}
}

27
skills/skill-adapter/assets/test-data.json Normal file
View File

@@ -0,0 +1,27 @@
{
"testCases": [
{
"name": "Basic activation test",
"input": "trigger phrase example",
"expected": {
"activated": true,
"toolsUsed": ["Read", "Grep"],
"success": true
}
},
{
"name": "Complex workflow test",
"input": "multi-step trigger example",
"expected": {
"activated": true,
"steps": 3,
"toolsUsed": ["Read", "Write", "Bash"],
"success": true
}
}
],
"fixtures": {
"sampleInput": "example data",
"expectedOutput": "processed result"
}
}

252
skills/skill-adapter/assets/webhook_handler_template.py Normal file
View File

@@ -0,0 +1,252 @@
#!/usr/bin/env python3
"""
A basic Python template for a webhook handler with signature verification
and idempotency. Provides a starting point for building robust and secure
webhook integrations.
"""
import hashlib
import hmac
import json
import logging
import os
import time
from functools import wraps
from http import HTTPStatus
from typing import Callable, Dict
from flask import Flask, request, jsonify
# Configure logging
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s')
# Environment variables (replace with your actual values)
WEBHOOK_SECRET = os.environ.get("WEBHOOK_SECRET", "your_secret_key")
MAX_RETRIES = int(os.environ.get("MAX_RETRIES", 3))
RETRY_DELAY = int(os.environ.get("RETRY_DELAY", 1)) # Initial delay in seconds
app = Flask(__name__)
class WebhookError(Exception):
"""Base class for webhook-related exceptions."""
pass
class SignatureVerificationError(WebhookError):
"""Raised when webhook signature verification fails."""
pass
class IdempotencyError(WebhookError):
"""Raised when idempotency check fails."""
pass
def verify_signature(request_data: bytes, signature: str, secret: str) -> None:
"""
Verifies the webhook signature against the request body and secret.
Args:
request_data: The raw bytes of the request body.
signature: The signature sent in the webhook request headers.
secret: The secret key used to generate the signature.
Raises:
SignatureVerificationError: If the signature does not match.
"""
try:
expected_signature = hmac.new(
secret.encode('utf-8'),
request_data,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(expected_signature, signature):
raise SignatureVerificationError("Invalid webhook signature.")
except Exception as e:
logging.error(f"Signature verification failed: {e}")
raise SignatureVerificationError("Signature verification failed.") from e
def idempotent(func: Callable) -> Callable:
"""
Decorator to ensure idempotency of webhook requests. This example uses a
simple in-memory store. For production, use a persistent database
(e.g., Redis, PostgreSQL).
Args:
func: The function to decorate (webhook handler).
Returns:
The decorated function.
Raises:
IdempotencyError: If the request ID has already been processed.
"""
processed_requests: Dict[str, bool] = {} # In-memory store (replace in production)
@wraps(func)
def wrapper(*args, **kwargs):
request_id = request.headers.get("X-Request-ID")
if not request_id:
logging.warning("Missing X-Request-ID header. Idempotency check skipped.")
return func(*args, **kwargs) # Skip idempotency check if no request ID
if request_id in processed_requests:
raise IdempotencyError(f"Request with ID {request_id} already processed.")
try:
result = func(*args, **kwargs)
processed_requests[request_id] = True # Mark as processed
return result
except Exception as e:
logging.error(f"Error processing request: {e}")
raise
return wrapper
def retry(func: Callable, max_retries: int = MAX_RETRIES, delay: int = RETRY_DELAY) -> Callable:
"""
Decorator to add retry logic with exponential backoff.
Args:
func: The function to decorate.
max_retries: The maximum number of retries.
delay: The initial delay in seconds.
Returns:
The decorated function.
"""
@wraps(func)
def wrapper(*args, **kwargs):
attempts = 0
while attempts < max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
attempts += 1
logging.warning(f"Attempt {attempts} failed: {e}. Retrying in {delay} seconds...")
time.sleep(delay)
delay *= 2 # Exponential backoff
logging.error(f"Max retries reached. Function {func.__name__} failed.")
raise
return wrapper
@app.route('/webhook', methods=['POST'])
@idempotent
@retry
def handle_webhook():
"""
Handles incoming webhook requests.
This function verifies the signature, processes the event, and returns a
success response. It also includes error handling and retry logic.
"""
signature = request.headers.get('X-Webhook-Signature')
if not signature:
logging.warning("Missing X-Webhook-Signature header.")
return jsonify({"error": "Missing signature"}), HTTPStatus.BAD_REQUEST
request_data = request.get_data()
try:
verify_signature(request_data, signature, WEBHOOK_SECRET)
except SignatureVerificationError as e:
logging.warning(f"Signature verification failed: {e}")
return jsonify({"error": str(e)}), HTTPStatus.UNAUTHORIZED
try:
payload = json.loads(request_data.decode('utf-8'))
event_type = payload.get("type") # Example: Get event type from payload
# Route the event to the appropriate handler (replace with your logic)
if event_type == "user.created":
process_user_created_event(payload)
elif event_type == "payment.succeeded":
process_payment_succeeded_event(payload)
else:
logging.warning(f"Unhandled event type: {event_type}")
return jsonify({"status": "unhandled"}), HTTPStatus.OK # Acknowledge the event
return jsonify({"status": "success"}), HTTPStatus.OK
except json.JSONDecodeError:
logging.error("Invalid JSON payload")
return jsonify({"error": "Invalid JSON payload"}), HTTPStatus.BAD_REQUEST
except Exception as e:
logging.exception("Error processing webhook")
return jsonify({"error": "Internal server error"}), HTTPStatus.INTERNAL_SERVER_ERROR
def process_user_created_event(payload: Dict) -> None:
"""
Processes a user.created event. This is a placeholder; replace with
your actual business logic.
Args:
payload: The event payload as a dictionary.
"""
user_id = payload.get("user_id")
logging.info(f"Processing user.created event for user ID: {user_id}")
# Add your business logic here (e.g., create user in your system)
time.sleep(0.1) # Simulate some processing time
def process_payment_succeeded_event(payload: Dict) -> None:
"""
Processes a payment.succeeded event. This is a placeholder; replace with
your actual business logic.
Args:
payload: The event payload as a dictionary.
"""
payment_id = payload.get("payment_id")
logging.info(f"Processing payment.succeeded event for payment ID: {payment_id}")
# Add your business logic here (e.g., update order status)
time.sleep(0.2) # Simulate some processing time
@app.errorhandler(SignatureVerificationError)
def handle_signature_error(error):
"""Handles SignatureVerificationError exceptions."""
return jsonify({"error": str(error)}), HTTPStatus.UNAUTHORIZED
@app.errorhandler(IdempotencyError)
def handle_idempotency_error(error):
"""Handles IdempotencyError exceptions."""
return jsonify({"error": str(error)}), HTTPStatus.CONFLICT
@app.errorhandler(Exception)
def handle_generic_error(error):
"""Handles generic exceptions."""
logging.exception("Unhandled exception")
return jsonify({"error": "Internal server error"}), HTTPStatus.INTERNAL_SERVER_ERROR
if __name__ == '__main__':
# Example Usage:
#
# 1. Set the WEBHOOK_SECRET environment variable.
# 2. Run the Flask app: python webhook_handler_template.py
# 3. Send a POST request to /webhook with a valid X-Webhook-Signature header.
#
# Example request:
#
# POST /webhook HTTP/1.1
# X-Webhook-Signature: <calculated_signature>
# X-Request-ID: <unique_request_id>
# Content-Type: application/json
#
# {
# "type": "user.created",
# "user_id": "123"
# }
app.run(debug=True, host='0.0.0.0', port=5000)