Webhook Approval Example

import requests

# Admin API endpoint (requires JWT authentication)
admin_api_url = "https://api.tracemem.com"
jwt_token = "your-jwt-token"

# Create webhook integration
response = requests.post(
    f"{admin_api_url}/v1/admin/integrations/webhook",
    headers={"Authorization": f"Bearer {jwt_token}"},
    json={
        "name": "production-approvals",
        "url": "https://your-server.com/tracemem/webhook",
        "events": ["approval.requested", "approval.resolved"],
        "timeout_ms": 5000,
        "max_retries": 3
    }
)

result = response.json()
webhook_id = result["webhook_id"]
outgoing_secret = result["outgoing_secret"]  # For verifying incoming requests
incoming_secret = result["incoming_secret"]  # For authenticating responses

# Store these securely!

# Create webhook credential
response = requests.post(
    f"{admin_api_url}/v1/admin/webhook-credentials",
    headers={"Authorization": f"Bearer {jwt_token}"},
    json={
        "name": "production-approvals",
        "expires_in": 31536000  # 1 year
    }
)

result = response.json()
credential_id = result["credential_id"]
webhook_secret = result["secret"]  # Use this to authenticate responses

# Store this securely!

#!/usr/bin/env python3
"""
TraceMem Webhook Approval Receiver

Receives approval requests from TraceMem, processes them,
and sends approval responses back.
"""

import os
import sys
import json
import hmac
import hashlib
import argparse
from datetime import datetime
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import queue
import requests


approval_queue = queue.Queue()


class ApprovalRequestHandler(BaseHTTPRequestHandler):
    """HTTP request handler for approval webhooks"""
    
    webhook_secret = None
    api_url = None
    
    def log_message(self, format, *args):
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        sys.stderr.write(f"[{timestamp}] {format % args}\n")
    
    def do_POST(self):
        """Handle POST requests"""
        if self.path != '/webhook':
            self.send_error(404, "Not Found")
            return
        
        # Read request body
        content_length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_length)
        
        # Parse JSON payload
        try:
            payload = json.loads(body.decode('utf-8'))
        except json.JSONDecodeError:
            self.send_error(400, "Invalid JSON")
            return
        
        # Validate event type
        if payload.get('event') != 'approval.requested':
            self.send_error(400, "Invalid event type")
            return
        
        # Add to queue for processing
        approval_queue.put({
            'payload': payload,
            'api_url': self.api_url,
            'webhook_secret': self.webhook_secret
        })
        
        # Send immediate success response
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.end_headers()
        self.wfile.write(json.dumps({"status": "queued"}).encode('utf-8'))


def approval_worker():
    """Worker thread that processes approval requests"""
    while True:
        item = approval_queue.get()
        payload = item['payload']
        api_url = item['api_url']
        webhook_secret = item['webhook_secret']
        
        try:
            process_approval(payload, api_url, webhook_secret)
        except Exception as e:
            print(f"Error processing approval: {e}")
        finally:
            approval_queue.task_done()


def process_approval(payload, api_url, webhook_secret):
    """Process approval request and prompt user (or make automatic decision)"""
    print("\n" + "="*80)
    print("APPROVAL REQUEST RECEIVED")
    print("="*80)
    print(f"Approval ID:  {payload.get('approval_id')}")
    print(f"Decision ID:  {payload.get('decision_id')}")
    print(f"Title:        {payload.get('title', 'N/A')}")
    print(f"Message:      {payload.get('message', 'N/A')}")
    print(f"Expires At:   {payload.get('expires_at', 'N/A')}")
    print("="*80)
    
    # Get approval decision (in production, this might be automatic or UI-based)
    decision = input("\nApprove or Reject? [approve/reject]: ").strip().lower()
    if decision not in ['approve', 'reject']:
        print("Invalid input. Defaulting to reject.")
        decision = 'reject'
    
    # Get rationale
    rationale = ""
    if payload.get('require_rationale'):
        rationale = input("Rationale (required): ").strip()
        if not rationale:
            print("Rationale is required. Defaulting to reject.")
            decision = 'reject'
            rationale = "No rationale provided"
    else:
        rationale = input("Rationale (optional): ").strip()
    
    # Send response to TraceMem API
    send_approval_response(
        api_url,
        webhook_secret,
        payload.get('token'),
        'approved' if decision == 'approve' else 'rejected',
        rationale
    )


def send_approval_response(api_url, webhook_secret, token, resolution, rationale):
    """Send approval response to TraceMem API"""
    url = f"{api_url}/v1/webhooks/approvals/respond"
    headers = {
        'Authorization': f'Bearer {webhook_secret}',
        'Content-Type': 'application/json',
    }
    
    payload = {
        'token': token,
        'resolution': resolution,
    }
    
    if rationale:
        payload['rationale'] = rationale
    
    print(f"\nSending {resolution} response to TraceMem API...")
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=10)
        
        if response.status_code == 200:
            print(f"✅ Approval response sent successfully!")
            result = response.json()
            print(f"   Approval ID: {result.get('approval_id')}")
            print(f"   Decision ID: {result.get('decision_id')}")
            print(f"   Status: {result.get('status')}")
        else:
            print(f"❌ Error: HTTP {response.status_code}")
            print(f"   Response: {response.text}")
    
    except requests.exceptions.RequestException as e:
        print(f"❌ Error sending approval response: {e}")
    
    print("="*80 + "\n")


def main():
    parser = argparse.ArgumentParser(description='TraceMem Webhook Approval Receiver')
    
    parser.add_argument(
        '--secret',
        help='Webhook credential secret',
        default=os.environ.get('TRACEMEM_WEBHOOK_SECRET')
    )
    
    parser.add_argument(
        '--api-url',
        help='TraceMem API base URL',
        default=os.environ.get('TRACEMEM_API_URL', 'https://api.tracemem.com')
    )
    
    parser.add_argument(
        '--port',
        type=int,
        default=int(os.environ.get('WEBHOOK_PORT', 9000)),
        help='Port to listen on (default: 9000)'
    )
    
    args = parser.parse_args()
    
    if not args.secret:
        print("Error: Webhook credential secret is required", file=sys.stderr)
        sys.exit(1)
    
    # Set class variables
    ApprovalRequestHandler.webhook_secret = args.secret
    ApprovalRequestHandler.api_url = args.api_url
    
    # Start approval worker thread
    worker_thread = threading.Thread(target=approval_worker, daemon=True)
    worker_thread.start()
    
    # Start HTTP server
    server_address = ('0.0.0.0', args.port)
    httpd = HTTPServer(server_address, ApprovalRequestHandler)
    
    print("="*80)
    print("TraceMem Webhook Approval Receiver")
    print("="*80)
    print(f"Listening on: http://0.0.0.0:{args.port}/webhook")
    print(f"API URL:      {args.api_url}")
    print("="*80)
    print("\nWaiting for approval requests... (Press Ctrl+C to stop)\n")
    
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n\nShutting down server...")
        httpd.shutdown()
        print("Server stopped.")


if __name__ == '__main__':
    main()

# Set environment variables
export TRACEMEM_WEBHOOK_SECRET="your-webhook-credential-secret"
export TRACEMEM_API_URL="https://api.tracemem.com"

# Run the server
python webhook_approval_receiver.py --port 9000