#!/usr/bin/env python3 """ Genesis Instantly Campaign Skill ================================= Skill for interacting with Instantly.ai API. Handles lead enrichment and campaign management. PM-037: Instantly Campaign Skill Enhancement - Supports: create_campaign, add_leads, pause/resume - Tracks open/reply rates - Complete campaign management """ import os import json import logging import requests from typing import List, Dict, Optional, Any from datetime import datetime, timedelta logger = logging.getLogger("InstantlySkill") logging.basicConfig(level=logging.INFO) class InstantlySkill: """ Skill for interacting with Instantly.ai API. Handles lead enrichment and campaign management. """ BASE_URL = "https://api.instantly.ai/api/v1" def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.getenv("INSTANTLY_API_KEY") if not self.api_key: logger.warning("Instantly API Key not found. Campaign features will be limited.") # Analytics tracking self.campaign_stats: Dict[str, Dict] = {} def _get_headers(self) -> Dict[str, str]: """Get authorization headers.""" if not self.api_key: raise ValueError("Instantly API Key is missing.") return { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } def _get_params(self) -> Dict[str, str]: """Get API key params for endpoints that use query params.""" return {"api_key": self.api_key} def _make_request(self, method: str, endpoint: str, data: Dict = None, params: Dict = None) -> Dict[str, Any]: """Make API request with error handling.""" url = f"{self.BASE_URL}/{endpoint}" # Merge API key into params if params is None: params = {} params["api_key"] = self.api_key try: if method.upper() == "GET": response = requests.get(url, params=params, timeout=30) elif method.upper() == "POST": response = requests.post(url, json=data, params=params, timeout=30) elif method.upper() == "PATCH": response = requests.patch(url, json=data, params=params, timeout=30) elif method.upper() == "DELETE": response = requests.delete(url, params=params, timeout=30) else: return {"error": f"Unsupported method: {method}"} if response.status_code in [200, 201]: return response.json() if response.text else {"status": "success"} elif response.status_code == 204: return {"status": "success"} else: logger.error(f"API Error {response.status_code}: {response.text}") return {"error": response.text, "status_code": response.status_code} except requests.exceptions.Timeout: return {"error": "timeout", "status_code": 408} except Exception as e: logger.error(f"Request failed: {e}") return {"error": str(e)} # ===== CAMPAIGN OPERATIONS ===== def create_campaign(self, name: str, sending_accounts: List[str], daily_limit: int = 50, schedule: Dict = None, subject_lines: List[str] = None, email_body: str = None) -> Dict[str, Any]: """ Creates a new email campaign. Args: name: Campaign name sending_accounts: List of email account IDs daily_limit: Daily sending limit per account schedule: Sending schedule configuration subject_lines: List of subject line variations email_body: Email template body Returns: Dict with campaign ID and status """ logger.info(f"Creating campaign: {name}") payload = { "name": name, "sending_accounts": sending_accounts, "daily_limit": daily_limit } if schedule: payload["schedule"] = schedule else: # Default schedule: weekdays 9am-5pm payload["schedule"] = { "timezone": "America/New_York", "days": ["mon", "tue", "wed", "thu", "fri"], "start_hour": 9, "end_hour": 17 } if subject_lines: payload["subject_lines"] = subject_lines if email_body: payload["email_body"] = email_body result = self._make_request("POST", "campaign/create", data=payload) if "campaign_id" in result: logger.info(f"Campaign created: {result['campaign_id']}") self.campaign_stats[result["campaign_id"]] = { "created_at": datetime.utcnow().isoformat(), "name": name, "leads_added": 0 } return result def get_campaign(self, campaign_id: str) -> Dict[str, Any]: """Gets campaign details.""" return self._make_request("GET", f"campaign/{campaign_id}") def list_campaigns(self, status: str = None) -> List[Dict[str, Any]]: """ Lists available campaigns. Args: status: Filter by status (active, paused, completed) Returns: List of campaign objects """ params = {} if status: params["status"] = status result = self._make_request("GET", "campaign/list", params=params) return result.get("campaigns", []) if isinstance(result, dict) else [] def update_campaign(self, campaign_id: str, updates: Dict[str, Any]) -> Dict[str, Any]: """Updates campaign settings.""" logger.info(f"Updating campaign: {campaign_id}") return self._make_request("PATCH", f"campaign/{campaign_id}", data=updates) def delete_campaign(self, campaign_id: str) -> Dict[str, Any]: """Deletes a campaign.""" logger.warning(f"Deleting campaign: {campaign_id}") return self._make_request("DELETE", f"campaign/{campaign_id}") # ===== CAMPAIGN STATUS OPERATIONS ===== def pause_campaign(self, campaign_id: str) -> Dict[str, Any]: """ Pauses an active campaign. Args: campaign_id: Target campaign ID Returns: Status result """ logger.info(f"Pausing campaign: {campaign_id}") return self._make_request("POST", "campaign/pause", data={"campaign_id": campaign_id}) def resume_campaign(self, campaign_id: str) -> Dict[str, Any]: """ Resumes a paused campaign. Args: campaign_id: Target campaign ID Returns: Status result """ logger.info(f"Resuming campaign: {campaign_id}") return self._make_request("POST", "campaign/resume", data={"campaign_id": campaign_id}) def start_campaign(self, campaign_id: str) -> Dict[str, Any]: """Starts a draft campaign.""" logger.info(f"Starting campaign: {campaign_id}") return self._make_request("POST", "campaign/start", data={"campaign_id": campaign_id}) # ===== LEAD OPERATIONS ===== def add_leads(self, campaign_id: str, leads: List[Dict], skip_duplicates: bool = True) -> Dict[str, Any]: """ Adds leads to a specific campaign. Args: campaign_id: Target campaign ID leads: List of lead objects with email, first_name, etc. skip_duplicates: Skip if email already exists Returns: Result with count of added leads """ logger.info(f"Adding {len(leads)} leads to campaign {campaign_id}") # Normalize lead format normalized_leads = [] for lead in leads: normalized_lead = { "email": lead.get("email"), "first_name": lead.get("first_name", lead.get("firstName", "")), "last_name": lead.get("last_name", lead.get("lastName", "")), "company_name": lead.get("company_name", lead.get("company", "")), "personalization": lead.get("personalization", "") } # Add any custom variables if "custom_variables" in lead: normalized_lead["custom_variables"] = lead["custom_variables"] normalized_leads.append(normalized_lead) payload = { "campaign_id": campaign_id, "leads": normalized_leads, "skip_if_in_workspace": skip_duplicates } result = self._make_request("POST", "lead/add", data=payload) if campaign_id in self.campaign_stats: self.campaign_stats[campaign_id]["leads_added"] += len(leads) return result def add_leads_from_csv(self, campaign_id: str, csv_path: str) -> Dict[str, Any]: """ Adds leads from a CSV file. Args: campaign_id: Target campaign ID csv_path: Path to CSV file Returns: Result with count of added leads """ import csv if not os.path.exists(csv_path): return {"error": f"File not found: {csv_path}"} leads = [] try: with open(csv_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) for row in reader: leads.append(row) logger.info(f"Loaded {len(leads)} leads from {csv_path}") return self.add_leads(campaign_id, leads) except Exception as e: logger.error(f"Failed to read CSV: {e}") return {"error": str(e)} def get_lead_status(self, campaign_id: str, email: str) -> Dict[str, Any]: """Gets the status of a specific lead.""" params = {"email": email} return self._make_request("GET", f"campaign/{campaign_id}/lead", params=params) def remove_lead(self, campaign_id: str, email: str) -> Dict[str, Any]: """Removes a lead from a campaign.""" return self._make_request("POST", "lead/remove", data={"campaign_id": campaign_id, "email": email}) # ===== ANALYTICS OPERATIONS ===== def get_campaign_analytics(self, campaign_id: str) -> Dict[str, Any]: """ Gets campaign analytics including open/reply rates. Args: campaign_id: Target campaign ID Returns: Dict with analytics data """ logger.info(f"Fetching analytics for campaign: {campaign_id}") result = self._make_request("GET", f"campaign/{campaign_id}/analytics") if "error" not in result: # Calculate rates sent = result.get("sent", 0) if sent > 0: result["open_rate"] = round((result.get("opened", 0) / sent) * 100, 2) result["reply_rate"] = round((result.get("replied", 0) / sent) * 100, 2) result["bounce_rate"] = round((result.get("bounced", 0) / sent) * 100, 2) result["click_rate"] = round((result.get("clicked", 0) / sent) * 100, 2) else: result["open_rate"] = 0 result["reply_rate"] = 0 result["bounce_rate"] = 0 result["click_rate"] = 0 return result def get_all_campaign_stats(self) -> Dict[str, Dict]: """Gets analytics for all campaigns.""" campaigns = self.list_campaigns() stats = {} for campaign in campaigns: campaign_id = campaign.get("id") if campaign_id: analytics = self.get_campaign_analytics(campaign_id) stats[campaign_id] = { "name": campaign.get("name"), "status": campaign.get("status"), **analytics } return stats def track_open_reply_rates(self, campaign_id: str) -> Dict[str, float]: """ Tracks open and reply rates for a campaign. Args: campaign_id: Target campaign ID Returns: Dict with open_rate and reply_rate """ analytics = self.get_campaign_analytics(campaign_id) return { "open_rate": analytics.get("open_rate", 0), "reply_rate": analytics.get("reply_rate", 0), "click_rate": analytics.get("click_rate", 0), "bounce_rate": analytics.get("bounce_rate", 0), "total_sent": analytics.get("sent", 0), "total_opened": analytics.get("opened", 0), "total_replied": analytics.get("replied", 0) } # ===== ACCOUNT OPERATIONS ===== def list_accounts(self) -> List[Dict[str, Any]]: """Lists sending accounts.""" result = self._make_request("GET", "account/list") return result.get("accounts", []) if isinstance(result, dict) else [] def get_account_warmup_status(self, email: str) -> Dict[str, Any]: """Gets warmup status for an account.""" return self._make_request("GET", f"account/{email}/warmup") # ===== ENRICHMENT OPERATIONS ===== def enrich_leads(self, csv_path: str, output_path: str) -> Dict[str, Any]: """ Enriches a CSV of leads using Instantly's enrichment endpoints. Args: csv_path: Input CSV path output_path: Output enriched CSV path Returns: Result with enrichment status """ if not self.api_key: return {"error": "API Key missing", "status": "failed"} if not os.path.exists(csv_path): return {"error": f"Input file not found: {csv_path}"} try: import csv with open(csv_path, 'r', encoding='utf-8') as f: reader = csv.DictReader(f) leads = list(reader) logger.info(f"Loaded {len(leads)} leads from {csv_path} for enrichment") # Enrich each lead enriched_leads = [] for lead in leads: # Call enrichment endpoint result = self._make_request("POST", "lead/enrich", data={"email": lead.get("email")}) if "error" not in result: lead.update(result.get("data", {})) enriched_leads.append(lead) # Write enriched data if enriched_leads: with open(output_path, 'w', encoding='utf-8', newline='') as f: writer = csv.DictWriter(f, fieldnames=enriched_leads[0].keys()) writer.writeheader() writer.writerows(enriched_leads) logger.info(f"Enriched {len(enriched_leads)} leads to {output_path}") return {"status": "success", "enriched_count": len(enriched_leads)} return {"status": "no_leads", "enriched_count": 0} except Exception as e: logger.error(f"Failed to enrich leads: {e}") return {"error": str(e)} # ===== UTILITY METHODS ===== def execute(self, action: str, **kwargs) -> Dict[str, Any]: """ Generic action executor for skill interface compatibility. Args: action: Action name **kwargs: Action-specific parameters Returns: Action result """ action_map = { "create_campaign": self.create_campaign, "get_campaign": self.get_campaign, "list_campaigns": self.list_campaigns, "update_campaign": self.update_campaign, "delete_campaign": self.delete_campaign, "pause": self.pause_campaign, "resume": self.resume_campaign, "start": self.start_campaign, "add_leads": self.add_leads, "add_leads_csv": self.add_leads_from_csv, "get_lead_status": self.get_lead_status, "remove_lead": self.remove_lead, "get_analytics": self.get_campaign_analytics, "track_rates": self.track_open_reply_rates, "list_accounts": self.list_accounts, "enrich_leads": self.enrich_leads } if action not in action_map: return {"error": f"Unknown action: {action}", "available": list(action_map.keys())} try: return action_map[action](**kwargs) except TypeError as e: return {"error": f"Invalid parameters for {action}: {e}"} except Exception as e: return {"error": str(e)} def get_health_status(self) -> Dict[str, Any]: """Returns skill health status.""" return { "api_key_configured": bool(self.api_key), "tracked_campaigns": len(self.campaign_stats), "status": "healthy" if self.api_key else "degraded" } if __name__ == "__main__": # Test stub skill = InstantlySkill() if skill.api_key: print("[TEST] Instantly Skill initialized with key.") # Test campaign listing campaigns = skill.execute("list_campaigns") print(f"[TEST] Campaigns: {campaigns}") # Test creating campaign result = skill.execute("create_campaign", name="Test Genesis Campaign", sending_accounts=["test@example.com"], subject_lines=["Hello {{first_name}}!"]) print(f"[TEST] Create Campaign: {result}") # Test analytics if campaigns: campaign_id = campaigns[0].get("id") rates = skill.execute("track_rates", campaign_id=campaign_id) print(f"[TEST] Open/Reply Rates: {rates}") # Health status print(f"[TEST] Health: {skill.get_health_status()}") else: print("[TEST] Instantly Skill initialized WITHOUT key (Mock Mode)") print(f"[TEST] Health: {skill.get_health_status()}")