Python Examples: Complete Integration Guide

Production-ready Python code for integrating FetchHook into your applications, scripts, and agents.

#Basic Fetch

Fetch webhooks from your mailbox using the requests library.

Simple Fetch

python

import requests
import os

API_KEY = os.getenv('FETCHHOOK_API_KEY')
SOURCE_ID = os.getenv('FETCHHOOK_SOURCE_ID')

response = requests.get(
    f'https://api.fetchhook.app/api/v1/{SOURCE_ID}',
    headers={'Authorization': f'Bearer {API_KEY}'}
)

webhooks = response.json()
print(f"Received {webhooks['count']} webhooks")

for event in webhooks['events']:
    print(f"Event ID: {event['event_id']}")
    print(f"Provider: {event['provider']}")
    print(f"Payload: {event['payload']}")

#Polling Loop

Continuously poll for new webhooks with exponential backoff and error handling.

Production Polling

python

import requests
import time
import os

API_KEY = os.getenv('FETCHHOOK_API_KEY')
SOURCE_ID = os.getenv('FETCHHOOK_SOURCE_ID')
POLL_INTERVAL = 5  # seconds

def fetch_webhooks():
    try:
        response = requests.get(
            f'https://api.fetchhook.app/api/v1/{SOURCE_ID}',
            headers={'Authorization': f'Bearer {API_KEY}'},
            timeout=10
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error fetching webhooks: {e}")
        return None

def process_webhook(event):
    """Process a single webhook event"""
    print(f"Processing {event['event_id']} from {event['provider']}")
    # Your business logic here
    
while True:
    data = fetch_webhooks()
    if data and data['count'] > 0:
        for event in data['events']:
            process_webhook(event)
    
    time.sleep(POLL_INTERVAL)

#Stripe Webhook Integration

Fetch and verify Stripe webhooks with signature validation.

Stripe + FetchHook

python

import requests
import stripe
import os

stripe.api_key = os.getenv('STRIPE_SECRET_KEY')
webhook_secret = os.getenv('STRIPE_WEBHOOK_SECRET')

API_KEY = os.getenv('FETCHHOOK_API_KEY')
SOURCE_ID = os.getenv('FETCHHOOK_SOURCE_ID')

response = requests.get(
    f'https://api.fetchhook.app/api/v1/{SOURCE_ID}',
    headers={'Authorization': f'Bearer {API_KEY}'}
)

for event_data in response.json()['events']:
    if event_data['provider'] != 'stripe':
        continue
    
    # Verify Stripe signature
    try:
        event = stripe.Webhook.construct_event(
            payload=event_data['payload'],
            sig_header=event_data['headers'].get('stripe-signature'),
            secret=webhook_secret
        )
        
        # Handle the event
        if event['type'] == 'payment_intent.succeeded':
            payment_intent = event['data']['object']
            print(f"Payment succeeded: {payment_intent['id']}")
            
    except stripe.error.SignatureVerificationError as e:
        print(f"Invalid signature: {e}")

#Async/Await with httpx

Use async Python for concurrent webhook fetching and processing.

Async Integration

python

import httpx
import asyncio
import os

API_KEY = os.getenv('FETCHHOOK_API_KEY')
SOURCE_ID = os.getenv('FETCHHOOK_SOURCE_ID')

async def fetch_webhooks():
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f'https://api.fetchhook.app/api/v1/{SOURCE_ID}',
            headers={'Authorization': f'Bearer {API_KEY}'}
        )
        return response.json()

async def process_webhook(event):
    # Async processing logic
    print(f"Processing {event['event_id']}")
    await asyncio.sleep(0.1)  # Simulate async work

async def main():
    while True:
        data = await fetch_webhooks()
        if data['count'] > 0:
            # Process webhooks concurrently
            tasks = [process_webhook(e) for e in data['events']]
            await asyncio.gather(*tasks)
        
        await asyncio.sleep(5)

if __name__ == "__main__":
    asyncio.run(main())

Agent Protocol Instruction

For production: Install dependencies with 'pip install requests stripe httpx', use environment variables for secrets, implement proper logging, add retry logic with exponential backoff, and handle webhook TTL (events expire after 24 hours).