The Growth Marketer's Guide to API-Driven Automation

import requests
from datetime import datetime

def monitor_hiring_signals(target_companies):
    """Monitor hiring spikes for outbound timing"""
    
    for company in target_companies:
        response = requests.get(
            f"https://api.predictleads.com/v1/signals/{company['domain']}",
            headers={"Authorization": f"Bearer {PREDICTLEADS_API_KEY}"}
        )
        
        signals = response.json()
        
        for signal in signals['hiring_spikes']:
            if signal['department'] == 'engineering' and signal['growth_rate'] > 30:
                # Trigger outbound sequence for expansion opportunity
                trigger_expansion_outreach(company, signal)
                
        for funding_event in signals['funding_rounds']:
            if funding_event['amount'] > 5000000:  # $5M+ rounds
                # Queue for immediate outreach
                priority_outreach_queue.add(company, funding_event)

def process_community_signals():
    """Identify warm leads from community engagement"""
    
    signals = commonroom_client.get_signals(
        platforms=['github', 'discord', 'linkedin'],
        activity_threshold=3,  # Multiple touchpoints
        recency_days=7
    )
    
    for signal in signals:
        # Score lead warmth based on engagement patterns
        warmth_score = calculate_engagement_score(signal)
        
        if warmth_score > 80:
            # Route to direct sales outreach
            create_high_priority_lead(signal)
        elif warmth_score > 50:
            # Add to nurture sequence
            add_to_nurture_campaign(signal)

def analyze_sales_conversations():
    """Extract insights from sales call transcripts"""
    
    calls = attention_client.get_recent_calls(days=7)
    
    for call in calls:
        # Identify competitor mentions
        competitors = extract_competitor_mentions(call['transcript'])
        
        # Track common objections
        objections = identify_objections(call['transcript'])
        
        # Update account record with conversation insights
        update_account_intelligence(call['account_id'], {
            'competitors_mentioned': competitors,
            'objections_raised': objections,
            'next_steps': extract_next_steps(call['transcript'])
        })

def enrich_contact_waterfall(linkedin_profile):
    """Cascade through multiple providers for verified contact data"""
    
    # Primary enrichment attempt
    contact_data = fullenrich_client.enrich_person(linkedin_profile)
    
    if contact_data['confidence'] > 90:
        return contact_data
    
    # Fallback to Prospeo for high-accuracy verification
    prospeo_result = prospeo_client.verify_email(contact_data['email'])
    
    if prospeo_result['deliverable']:
        contact_data['verified'] = True
        return contact_data
    
    # Final fallback through Apollo technographic data
    apollo_enrichment = apollo_client.enrich_person(
        linkedin_url=linkedin_profile,
        include_technographics=True
    )
    
    return merge_enrichment_data(contact_data, apollo_enrichment)

def linkedin_to_contact_pipeline(linkedin_urls):
    """Convert LinkedIn profiles to verified contact records"""
    
    enriched_contacts = []
    
    for url in linkedin_urls:
        # Extract contact data from LinkedIn profile
        contact = wiza_client.extract_contact(url)
        
        # Verify email deliverability
        verification = verify_email_deliverability(contact['email'])
        
        if verification['deliverable']:
            # Enrich with additional firmographic data
            company_data = apollo_client.enrich_company(contact['company'])
            
            enriched_contacts.append({
                **contact,
                **company_data,
                'verified': True,
                'enriched_at': datetime.now()
            })
    
    return enriched_contacts

def scrape_competitor_customers():
    """Mine competitor customer lists for expansion opportunities"""
    
    # Scrape competitor case studies and customer testimonials
    competitor_customers = apify_client.run_actor(
        "competitor-customer-scraper",
        run_input={
            "competitor_domains": ["competitor1.com", "competitor2.com"],
            "scrape_depth": "customer-pages"
        }
    )
    
    prospects = []
    for customer in competitor_customers:
        # Enrich with contact data
        contacts = find_decision_makers(customer['company_domain'])
        
        for contact in contacts:
            prospects.append({
                'company': customer['company_name'],
                'contact': contact,
                'competitive_intel': customer['use_case'],
                'source': 'competitor_analysis'
            })
    
    return prospects

def launch_signal_based_campaign(prospects, signal_type):
    """Launch personalized campaigns based on detected signals"""
    
    # Select email template based on signal
    template_map = {
        'hiring_spike': 'expansion-opportunity-template',
        'funding_round': 'growth-congratulations-template',
        'competitor_mention': 'competitive-alternative-template'
    }
    
    template = template_map.get(signal_type, 'general-outreach-template')
    
    campaign = instantly_client.create_campaign(
        name=f"Signal-Based Outreach - {signal_type}",
        template=template,
        prospects=prospects,
        settings={
            'daily_limit': 50,
            'warmup_enabled': True,
            'reply_tracking': True,
            'webhook_url': 'https://your-app.com/webhooks/email-reply'
        }
    )
    
    return campaign

def create_abm_retargeting_campaign(target_accounts):
    """Build LinkedIn ads targeting specific accounts"""
    
    # Create matched audience from target account list
    audience = linkedin_ads_client.create_matched_audience(
        name="ABM Target Accounts",
        account_list=[account['linkedin_url'] for account in target_accounts]
    )
    
    # Launch awareness campaign
    campaign = linkedin_ads_client.create_campaign(
        name="ABM Awareness Campaign",
        targeting={
            'matched_audiences': [audience['id']],
            'job_functions': ['marketing', 'sales', 'engineering']
        },
        budget={
            'amount': 5000,
            'currency': 'USD',
            'type': 'monthly'
        },
        creative={
            'headline': 'Built for Companies Like Yours',
            'description': 'See how similar companies achieve results',
            'call_to_action': 'Learn More'
        }
    )
    
    return campaign

def route_multichannel_sequence(prospect):
    """Route prospects to optimal outreach channels"""
    
    # Determine best channel based on enrichment data
    if prospect['seniority_level'] == 'C-Level':
        sequence_type = 'executive-outreach'
        channels = ['email', 'linkedin', 'phone']
    elif prospect['department'] == 'engineering':
        sequence_type = 'technical-outreach'  
        channels = ['email', 'linkedin']
    else:
        sequence_type = 'general-outreach'
        channels = ['email']
    
    # Launch sequence with personalized messaging
    sequence = lemlist_client.add_to_sequence(
        sequence_name=sequence_type,
        prospect=prospect,
        personalization={
            'company_signal': prospect.get('trigger_signal'),
            'use_case': prospect.get('potential_use_case'),
            'competitor_mentioned': prospect.get('current_solution')
        }
    )
    
    return sequence

def build_full_cycle_workflow():
    """Complete signal-to-revenue workflow"""
    
    workflow = {
        'trigger': {
            'type': 'webhook',
            'source': 'predictleads_hiring_signal'
        },
        'steps': [
            {
                'name': 'enrich_contact',
                'service': 'fullenrich',
                'action': 'cascade_enrichment'
            },
            {
                'name': 'verify_deliverability',
                'service': 'prospeo',
                'action': 'verify_email'
            },
            {
                'name': 'route_outreach',
                'service': 'lemlist',
                'action': 'add_to_sequence',
                'conditions': {
                    'if_verified': 'high_priority_sequence',
                    'if_unverified': 'linkedin_only_sequence'
                }
            },
            {
                'name': 'create_crm_record',
                'service': 'attio',
                'action': 'create_person',
                'data_mapping': {
                    'enrichment_data': '$node["enrich_contact"].json',
                    'signal_context': '$node["Webhook"].json["signal_details"]'
                }
            }
        ]
    }
    
    return n8n_client.deploy_workflow(workflow)
      
  

def create_response_routing_agent():
    """AI agent that routes email replies to appropriate actions"""
    
    agent_config = {
        'name': 'Email Response Router',
        'model': 'claude-3.5-sonnet',
        'prompt': '''
        Analyze this email reply and determine the appropriate next action:
        
        1. Positive interest - Schedule demo
        2. Objection - Route to objection handling sequence  
        3. Out of office - Pause sequence, retry in specified timeframe
        4. Not interested - Mark as unqualified, remove from sequences
        5. Referral - Extract referral contact info, add to pipeline
        
        Return structured JSON with action and reasoning.
        ''',
        'tools': [
            'cal_com_scheduling',
            'attio_crm_update',
            'lemlist_sequence_management'
        ]
    }
    
    return relevance_ai_client.create_agent(agent_config)

def sync_enrichment_to_crm(person_id, enrichment_data):
    """Update CRM records with enriched data and trigger workflows"""
    
    # Update person record with enriched data
    updated_record = attio_client.update_person(
        person_id=person_id,
        attributes={
            'email_verified': enrichment_data['email_verified'],
            'phone_number': enrichment_data['phone'],
            'seniority_level': enrichment_data['seniority'],
            'department': enrichment_data['department'],
            'technologies_used': enrichment_data['tech_stack'],
            'company_size': enrichment_data['company_employee_count'],
            'last_enriched': datetime.now()
        }
    )
    
    # Trigger lead scoring update
    score = calculate_lead_score(enrichment_data)
    
    # Auto-assign to appropriate sales rep based on territory/vertical
    assigned_rep = route_to_sales_rep(enrichment_data)
    
    # Create follow-up task if high-value prospect
    if score > 80:
        attio_client.create_task(
            title=f"High-value prospect: {enrichment_data['full_name']}",
            assignee=assigned_rep,
            due_date=datetime.now() + timedelta(days=1),
            details=f"Score: {score}, Signal: {enrichment_data['trigger_signal']}"
        )
    
    return updated_record

def handle_crm_webhooks(webhook_data):
    """Process real-time CRM changes and trigger appropriate actions"""
    
    event_type = webhook_data['event_type']
    
    if event_type == 'person.email_replied':
        # Route email reply through AI agent for response classification
        route_email_response(webhook_data['person_id'], webhook_data['email_content'])
        
    elif event_type == 'deal.stage_changed':
        if webhook_data['new_stage'] == 'demo_scheduled':
            # Trigger demo preparation workflow
            prepare_demo_materials(webhook_data['deal_id'])
            
        elif webhook_data['new_stage'] == 'closed_won':
            # Update customer success and billing systems
            trigger_onboarding_sequence(webhook_data['deal_id'])
            
    elif event_type == 'person.score_updated':
        if webhook_data['new_score'] > 90:
            # Auto-create high-priority task for sales rep
            create_urgent_follow_up_task(webhook_data['person_id'])

def create_intelligent_booking_flow(prospect_data):
    """Create personalized booking experience based on prospect intelligence"""
    
    # Determine meeting type based on company characteristics
    if prospect_data['company_size'] > 1000:
        meeting_type = 'enterprise-demo'
        duration = 45
    elif prospect_data['technical_role']:
        meeting_type = 'technical-deep-dive'
        duration = 30
    else:
        meeting_type = 'standard-demo'
        duration = 30
    
    # Create personalized booking link
    booking_link = cal_client.create_booking_link(
        meeting_type=meeting_type,
        duration=duration,
        pre_fill={
            'name': prospect_data['full_name'],
            'email': prospect_data['email'],
            'company': prospect_data['company_name']
        },
        custom_questions=[
            {
                'question': 'What\'s your biggest challenge with [current solution]?',
                'required': True,
                'type': 'text'
            },
            {
                'question': 'How many [relevant metric] do you currently handle?',
                'required': False,
                'type': 'number'
            }
        ]
    )
    
    # Set up automatic CRM sync for booking
    cal_client.configure_webhook(
        booking_link['id'],
        webhook_url='https://your-app.com/webhooks/meeting-booked',
        events=['booking.created', 'booking.rescheduled', 'booking.cancelled']
    )
    
    return booking_link

def track_revenue_attribution():
    """Connect every touchpoint from signal to closed revenue"""
    
    # Get attribution data for closed deals
    attribution_data = dreamdata_client.get_attribution_analysis(
        date_range='last_30_days',
        attribution_model='time_decay'
    )
    
    for deal in attribution_data['deals']:
        # Identify highest-value signal sources
        top_signals = analyze_signal_influence(deal['touchpoints'])
        
        # Calculate ROI by channel and signal type
        channel_roi = calculate_channel_roi(deal['touchpoints'], deal['value'])
        
        # Update signal detection thresholds based on conversion data
        optimize_signal_thresholds(top_signals, deal['conversion_time'])
        
        # Feed insights back to targeting algorithms
        update_targeting_models(deal['account_characteristics'], deal['outcome'])

def automate_billing_workflows():
    """Handle subscription lifecycle and usage-based billing"""
    
    # Set up usage-based billing for API consumption
    hyperline_client.configure_usage_billing(
        product_id='api-usage-tier',
        usage_events=[
            'api_call_made',
            'data_enrichment_request', 
            'email_sent',
            'meeting_booked'
        ],
        billing_cycle='monthly',
        webhooks=[
            'subscription.created',
            'payment.succeeded',
            'payment.failed'
        ]
    )
    
    # Auto-upgrade customers based on usage patterns
    def handle_usage_webhook(usage_data):
        if usage_data['monthly_api_calls'] > usage_data['plan_limit'] * 0.8:
            # Trigger upgrade conversation
            create_expansion_opportunity(usage_data['customer_id'])

class ResilientGTMPipeline:
    def __init__(self):
        self.max_retries = 3
        self.fallback_providers = {
            'email_enrichment': ['fullenrich', 'prospeo', 'apollo'],
            'email_sending': ['instantly', 'lemlist'],
            'crm_update': ['attio', 'hubspot_backup']
        }
    
    def execute_with_fallback(self, action, data, service_type):
        """Execute action with automatic fallback to alternative providers"""
        
        providers = self.fallback_providers.get(service_type, [])
        
        for provider in providers:
            for attempt in range(self.max_retries):
                try:
                    result = self.execute_action(provider, action, data)
                    
                    # Log successful execution
                    self.log_success(provider, action, attempt + 1)
                    return result
                    
                except RateLimitError:
                    # Wait and retry with exponential backoff
                    wait_time = (2 ** attempt) * 60  # 1min, 2min, 4min
                    time.sleep(wait_time)
                    
                except ProviderError as e:
                    # Log error and try next provider
                    self.log_provider_error(provider, action, e)
                    break
                    
        # If all providers failed, escalate to human review
        self.escalate_to_human(action, data, "All providers failed")
        
    def monitor_pipeline_health(self):
        """Monitor system performance and automatically adjust"""
        
        metrics = self.get_pipeline_metrics()
        
        # Adjust rate limits based on recent errors
        if metrics['error_rate'] > 0.1:  # 10% error rate
            self.reduce_rate_limits(factor=0.8)
            
        # Switch primary providers if success rates drop
        for service, providers in self.fallback_providers.items():
            success_rates = self.calculate_success_rates(providers)
            
            if success_rates[providers[0]] < 0.9:  # Primary below 90%
                # Promote better-performing provider to primary
                self.reorder_providers(service, success_rates)

def calculate_gtm_machine_metrics():
    """Comprehensive metrics for automated GTM performance"""
    
    metrics = {
        # Signal detection performance
        'signal_accuracy': calculate_signal_conversion_rate(),
        'signal_volume': count_signals_by_type(last_30_days),
        'signal_to_contact_rate': calculate_enrichment_success(),
        
        # Outreach effectiveness  
        'email_deliverability': get_email_delivery_rates(),
        'response_rates_by_channel': calculate_channel_response_rates(),
        'meeting_booking_conversion': get_booking_conversion_rates(),
        
        # Revenue attribution
        'signal_to_revenue': calculate_signal_roi(),
        'channel_attribution': get_revenue_by_channel(),
        'time_to_close': calculate_average_sales_cycle(),
        
        # System reliability
        'api_uptime': calculate_provider_reliability(),
        'error_rates': get_pipeline_error_rates(),
        'data_quality_score': assess_enrichment_accuracy()
    }
    
    # Identify optimization opportunities
    optimization_insights = analyze_performance_patterns(metrics)
    
    return {
        'metrics': metrics,
        'insights': optimization_insights,
        'recommended_actions': generate_optimization_recommendations(metrics)
    }