""" Instantly.ai API V2 Client ========================== Python wrapper for the Instantly.ai cold email platform API V2. API Docs: https://developer.instantly.ai/api/v2 Auth: Bearer token Base URL: https://api.instantly.ai/api/v2 Created: 2026-02-17 """ import json import logging from typing import Any, Dict, List, Optional import requests logger = logging.getLogger(__name__) # Instantly API V2 Base URL BASE_URL = "https://api.instantly.ai/api/v2" class InstantlyAPIError(Exception): """Raised when the Instantly API returns an error.""" def __init__(self, status_code: int, message: str, response_body: Any = None): self.status_code = status_code self.message = message self.response_body = response_body super().__init__(f"Instantly API Error {status_code}: {message}") class InstantlyClient: """ Client for the Instantly.ai API V2. Usage: client = InstantlyClient(api_key="your_api_key_here") campaigns = client.list_campaigns() client.create_lead(campaign_id="...", email="...", first_name="...") """ def __init__(self, api_key: str, timeout: int = 30): """ Initialize the Instantly API client. Args: api_key: Instantly API V2 key (Bearer token) timeout: Request timeout in seconds """ self.api_key = api_key self.timeout = timeout self.session = requests.Session() self.session.headers.update({ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Accept": "application/json", }) def _request( self, method: str, endpoint: str, params: Optional[Dict] = None, json_data: Optional[Dict] = None, extra_timeout: Optional[int] = None, ) -> Any: """ Make a request to the Instantly API. Args: method: HTTP method (GET, POST, PATCH, DELETE) endpoint: API endpoint path (e.g., "/campaigns") params: Query parameters json_data: JSON request body extra_timeout: Override default timeout Returns: Parsed JSON response Raises: InstantlyAPIError: If the API returns an error """ url = f"{BASE_URL}{endpoint}" timeout = extra_timeout or self.timeout logger.debug(f"Instantly API {method} {url}") if json_data: logger.debug(f" Body: {json.dumps(json_data, indent=2)}") try: response = self.session.request( method=method, url=url, params=params, json=json_data, timeout=timeout, ) except requests.exceptions.Timeout: raise InstantlyAPIError(408, f"Request to {url} timed out after {timeout}s") except requests.exceptions.ConnectionError as e: raise InstantlyAPIError(0, f"Connection error to {url}: {e}") # Log rate limit headers if present rate_remaining = response.headers.get("X-RateLimit-Remaining") rate_limit = response.headers.get("X-RateLimit-Limit") if rate_remaining is not None: logger.debug(f" Rate limit: {rate_remaining}/{rate_limit} remaining") if response.status_code >= 400: try: error_body = response.json() except (json.JSONDecodeError, ValueError): error_body = response.text error_msg = error_body if isinstance(error_body, dict): error_msg = error_body.get("message", error_body.get("error", str(error_body))) raise InstantlyAPIError(response.status_code, str(error_msg), error_body) # Some endpoints return empty responses (204 No Content) if response.status_code == 204 or not response.text.strip(): return {"status": "success", "status_code": response.status_code} try: return response.json() except (json.JSONDecodeError, ValueError): return {"raw_text": response.text, "status_code": response.status_code} # ========================================================================= # CAMPAIGN ENDPOINTS # ========================================================================= def list_campaigns( self, limit: int = 10, status: Optional[int] = None, search: Optional[str] = None, starting_after: Optional[str] = None, ) -> Dict: """ List campaigns. GET /campaigns Args: limit: Number of campaigns to return (default 10) status: Filter by status (0=draft, 1=active, 2=paused, 3=completed) search: Search campaigns by name starting_after: Pagination cursor Returns: Dict with campaigns list and pagination info """ params = {"limit": limit} if status is not None: params["status"] = status if search: params["search"] = search if starting_after: params["starting_after"] = starting_after return self._request("GET", "/campaigns", params=params) def get_campaign(self, campaign_id: str) -> Dict: """ Get a specific campaign by ID. GET /campaigns/{id} Args: campaign_id: The campaign UUID Returns: Campaign details dict """ return self._request("GET", f"/campaigns/{campaign_id}") def create_campaign( self, name: str, sequences: Optional[List[Dict]] = None, campaign_schedule: Optional[Dict] = None, ) -> Dict: """ Create a new campaign. POST /campaigns Args: name: Campaign name sequences: List of sequence objects with steps and variants. Only the first element is used. Format: [{"steps": [{"type": "email", "delay": 0, "variants": [{"subject": "...", "body": "..."}]}]}] campaign_schedule: Schedule config with timezone, start_date, end_date, schedules Returns: Created campaign dict with id """ body: Dict[str, Any] = {"name": name} if sequences: body["sequences"] = sequences if campaign_schedule: body["campaign_schedule"] = campaign_schedule return self._request("POST", "/campaigns", json_data=body) def update_campaign(self, campaign_id: str, **kwargs) -> Dict: """ Update a campaign. PATCH /campaigns/{id} Args: campaign_id: Campaign UUID **kwargs: Fields to update (name, sequences, campaign_schedule, etc.) Returns: Updated campaign dict """ return self._request("PATCH", f"/campaigns/{campaign_id}", json_data=kwargs) def activate_campaign(self, campaign_id: str) -> Dict: """ Activate (start) a campaign. POST /campaigns/{id}/activate Args: campaign_id: Campaign UUID Returns: Activation result """ return self._request("POST", f"/campaigns/{campaign_id}/activate", json_data={}) def pause_campaign(self, campaign_id: str) -> Dict: """ Pause a campaign. POST /campaigns/{id}/pause Args: campaign_id: Campaign UUID Returns: Pause result """ return self._request("POST", f"/campaigns/{campaign_id}/pause", json_data={}) def delete_campaign(self, campaign_id: str) -> Dict: """ Delete a campaign. WARNING: This is permanent. DELETE /campaigns/{id} Args: campaign_id: Campaign UUID Returns: Deletion result """ return self._request("DELETE", f"/campaigns/{campaign_id}") # ========================================================================= # LEAD ENDPOINTS # ========================================================================= def create_lead( self, email: str, campaign_id: Optional[str] = None, list_id: Optional[str] = None, first_name: Optional[str] = None, last_name: Optional[str] = None, company_name: Optional[str] = None, website: Optional[str] = None, phone: Optional[str] = None, personalization: Optional[str] = None, custom_variables: Optional[Dict[str, Any]] = None, ) -> Dict: """ Create a single lead. POST /leads Args: email: Lead email address (required) campaign_id: Campaign UUID to add lead to list_id: Lead list UUID to add lead to first_name: Lead first name last_name: Lead last name company_name: Lead company name website: Lead website URL phone: Lead phone number personalization: Custom personalization text custom_variables: Dict of custom variable key-value pairs. Values must be string, number, boolean, or null. Objects/arrays are NOT allowed. Returns: Created lead dict """ body: Dict[str, Any] = {"email": email} if campaign_id: body["campaign_id"] = campaign_id if list_id: body["list_id"] = list_id if first_name: body["first_name"] = first_name if last_name: body["last_name"] = last_name if company_name: body["company_name"] = company_name if website: body["website"] = website if phone: body["phone"] = phone if personalization: body["personalization"] = personalization if custom_variables: body["custom_variables"] = custom_variables return self._request("POST", "/leads", json_data=body) def add_leads_bulk( self, leads: List[Dict], campaign_id: Optional[str] = None, list_id: Optional[str] = None, skip_if_in_workspace: bool = False, skip_if_in_campaign: bool = True, ) -> Dict: """ Add leads in bulk to a campaign or list. POST /leads/batch Args: leads: List of lead dicts (max 1000). Each must have at minimum 'email'. campaign_id: Campaign UUID to add leads to list_id: Lead list UUID to add leads to skip_if_in_workspace: Skip if lead exists in any workspace campaign skip_if_in_campaign: Skip if already in the target campaign (default True) Returns: Bulk operation result """ body: Dict[str, Any] = { "leads": leads, "skip_if_in_workspace": skip_if_in_workspace, "skip_if_in_campaign": skip_if_in_campaign, } if campaign_id: body["campaign_id"] = campaign_id if list_id: body["list_id"] = list_id return self._request("POST", "/leads/batch", json_data=body, extra_timeout=60) def list_leads( self, campaign_id: Optional[str] = None, list_id: Optional[str] = None, limit: int = 10, email: Optional[str] = None, starting_after: Optional[str] = None, ) -> Dict: """ List leads with optional filters. GET /leads/search NOTE: The V2 API uses /leads/search as the list endpoint, not /leads. This endpoint requires at least a campaign_id or list_id. Args: campaign_id: Filter by campaign list_id: Filter by lead list limit: Number of results email: Filter by email address starting_after: Pagination cursor Returns: Dict with leads list """ params: Dict[str, Any] = {"limit": limit} if campaign_id: params["campaign_id"] = campaign_id if list_id: params["list_id"] = list_id if email: params["email"] = email if starting_after: params["starting_after"] = starting_after return self._request("GET", "/leads/search", params=params) def get_lead(self, lead_id: str) -> Dict: """ Get a specific lead by ID. GET /leads/{id} Args: lead_id: Lead UUID Returns: Lead details dict """ return self._request("GET", f"/leads/{lead_id}") def update_lead(self, lead_id: str, **kwargs) -> Dict: """ Update a lead. PATCH /leads/{id} Args: lead_id: Lead UUID **kwargs: Fields to update Returns: Updated lead dict """ return self._request("PATCH", f"/leads/{lead_id}", json_data=kwargs) def delete_lead(self, lead_id: str) -> Dict: """ Delete a lead. WARNING: This is permanent. DELETE /leads/{id} Args: lead_id: Lead UUID Returns: Deletion result """ return self._request("DELETE", f"/leads/{lead_id}") # ========================================================================= # EMAIL ENDPOINTS # ========================================================================= def list_emails( self, campaign_id: Optional[str] = None, lead_id: Optional[str] = None, limit: int = 10, ) -> Dict: """ List sent/received emails. GET /emails Args: campaign_id: Filter by campaign lead_id: Filter by lead limit: Number of results Returns: Dict with emails list """ params: Dict[str, Any] = {"limit": limit} if campaign_id: params["campaign_id"] = campaign_id if lead_id: params["lead_id"] = lead_id return self._request("GET", "/emails", params=params) # ========================================================================= # ACCOUNT ENDPOINTS # ========================================================================= def list_accounts(self, limit: int = 10) -> Dict: """ List sending accounts. GET /accounts Args: limit: Number of results Returns: Dict with accounts list """ return self._request("GET", "/accounts", params={"limit": limit}) # ========================================================================= # ACCOUNT-CAMPAIGN MAPPING # ========================================================================= def get_account_campaign_mappings(self, email: str) -> Dict: """ Get campaigns associated with a sending account email. GET /account-campaign-mappings/{email} NOTE: The V2 API only supports GET for account-campaign mappings. Assigning accounts to campaigns must be done via the Instantly dashboard. Args: email: Sending account email address Returns: List of campaign mappings for this account """ return self._request("GET", f"/account-campaign-mappings/{email}") def get_default_client() -> InstantlyClient: """ Create a client using the Genesis API key. Returns: Configured InstantlyClient instance """ API_KEY = "MjBjODUxNGYtNjA5MC00NjY4LWFhY2UtOWZmYTE3NDhhMmQ1OnRTYkpqRU94RHJveg==" return InstantlyClient(api_key=API_KEY) # ============================================================================ # CLI for testing # ============================================================================ if __name__ == "__main__": import sys logging.basicConfig(level=logging.INFO) client = get_default_client() if len(sys.argv) < 2: print("Usage: python instantly_client.py [args...]") print("Commands:") print(" list-campaigns List all campaigns") print(" get-campaign Get campaign details") print(" list-accounts List sending accounts") print(" list-leads List leads in campaign") sys.exit(1) command = sys.argv[1] if command == "list-campaigns": result = client.list_campaigns(limit=20) print(json.dumps(result, indent=2)) elif command == "get-campaign": if len(sys.argv) < 3: print("Error: campaign_id required") sys.exit(1) result = client.get_campaign(sys.argv[2]) print(json.dumps(result, indent=2)) elif command == "list-accounts": result = client.list_accounts(limit=20) print(json.dumps(result, indent=2)) elif command == "list-leads": if len(sys.argv) < 3: print("Error: campaign_id required") sys.exit(1) result = client.list_leads(campaign_id=sys.argv[2]) print(json.dumps(result, indent=2)) else: print(f"Unknown command: {command}") sys.exit(1)