Python Examples
Python
Complete Python examples for integrating with the Magnetite API using requests
Installation
Install the required dependencies:
pip install requests python-dotenvBasic Client Class
Create a reusable client class for interacting with the Magnetite API:
# magnetite_client.py
import os
import requests
import time
from typing import Dict, Any, List, Optional
from dotenv import load_dotenv
load_dotenv()
class MagnetiteClient:
def __init__(self, api_key: str = None):
self.api_key = api_key or os.getenv('MAGNETITE_API_KEY')
self.base_url = 'https://magnetite.ai/api'
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {self.api_key}',
'Content-Type': 'application/json'
})
if not self.api_key:
raise ValueError("API key is required. Set MAGNETITE_API_KEY environment variable or pass it directly.")
def add_lead(self, campaign_id: str, lead_data: Dict[str, Any]) -> Dict[str, Any]:
"""Add a lead to a campaign for automated outreach."""
try:
response = self.session.post(
f'{self.base_url}/campaigns/{campaign_id}/leads',
json=lead_data
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
self._handle_error(e)
def add_leads_bulk(self, campaign_id: str, leads: List[Dict[str, Any]], delay: float = 1.0) -> List[Dict[str, Any]]:
"""Add multiple leads with rate limiting."""
results = []
for i, lead in enumerate(leads):
try:
result = self.add_lead(campaign_id, lead)
results.append({
'lead': lead,
'success': True,
'result': result
})
print(f"Added lead {i+1}/{len(leads)}: {lead.get('email')}")
except Exception as e:
results.append({
'lead': lead,
'success': False,
'error': str(e)
})
print(f"Failed to add lead {lead.get('email')}: {e}")
# Respect rate limits
if i < len(leads) - 1:
time.sleep(delay)
return results
def _handle_error(self, error):
"""Handle and format API errors."""
if hasattr(error, 'response') and error.response is not None:
try:
error_data = error.response.json()
error_msg = error_data.get('error', 'Unknown API error')
raise Exception(f"API Error ({error.response.status_code}): {error_msg}")
except ValueError:
raise Exception(f"HTTP Error ({error.response.status_code}): {error.response.text}")
else:
raise Exception(f"Request Error: {str(error)}")
Environment Configuration
Create a .env file for your credentials:
# .env
MAGNETITE_API_KEY=sk_your_api_key_here
MAGNETITE_CAMPAIGN_ID=your_campaign_id_hereComplete Examples
Add a Single Lead
Simple example of adding a lead to your campaign
# add_lead.py
import os
from magnetite_client import MagnetiteClient
def add_single_lead():
client = MagnetiteClient()
try:
result = client.add_lead(
os.getenv('MAGNETITE_CAMPAIGN_ID'),
{
'companyName': 'Acme Corp',
'contactName': 'John Smith',
'email': 'john@acme.com',
'title': 'VP of Sales',
'domain': 'acme.com',
'linkedInUrl': 'https://linkedin.com/in/johnsmith',
'notes': 'Met at SaaStr conference'
}
)
print("Lead added successfully!")
print(f"Prospect ID: {result['prospectId']}")
print(f"Status: {result['status']}")
print(f"Message: {result['message']}")
return result
except Exception as e:
print(f"Error adding lead: {e}")
raise
if __name__ == "__main__":
add_single_lead()
print("Done!")Add Lead with Enrichment Data
Pass pre-enriched company and contact data
# add_enriched_lead.py
import os
from magnetite_client import MagnetiteClient
def add_enriched_lead():
client = MagnetiteClient()
try:
result = client.add_lead(
os.getenv('MAGNETITE_CAMPAIGN_ID'),
{
# Required fields
'companyName': 'TechCorp Inc',
'contactName': 'Jane Doe',
'email': 'jane@techcorp.com',
# Optional basic fields
'title': 'Marketing Director',
'domain': 'techcorp.com',
'phone': '+1-555-123-4567',
'linkedInUrl': 'https://linkedin.com/in/janedoe',
'companyLinkedInUrl': 'https://linkedin.com/company/techcorp',
'notes': 'Referred by partner XYZ',
# Pre-enriched company data (optional)
'company': {
'industry': 'Technology',
'employeeCount': 250,
'revenue': '$50M-100M',
'fundingStage': 'Series B',
'technologies': ['React', 'Node.js', 'AWS'],
'description': 'Enterprise SaaS platform'
},
# Pre-enriched contact data (optional)
'contact': {
'bio': 'Marketing leader with 10+ years experience',
'seniority': 'Director',
'department': 'Marketing',
'city': 'San Francisco',
'state': 'CA',
'country': 'USA'
},
# Extra context for research agent (optional)
'extraContext': {
'meetingNotes': 'Discussed integration needs',
'painPoints': ['Lead generation', 'Sales automation'],
'referredBy': 'Partner program'
}
}
)
print("Enriched lead added!")
print(f"Result: {result}")
except Exception as e:
print(f"Error: {e}")
raise
if __name__ == "__main__":
add_enriched_lead()
print("Done!")Bulk Lead Import
Import multiple leads with rate limiting
# bulk_import.py
import os
from magnetite_client import MagnetiteClient
def bulk_import():
client = MagnetiteClient()
leads = [
{
'companyName': 'StartupCo',
'contactName': 'Sarah Johnson',
'email': 'sarah@startupco.com',
'title': 'CEO'
},
{
'companyName': 'BigCorp Industries',
'contactName': 'Michael Chen',
'email': 'michael@bigcorp.com',
'title': 'Sales Director'
},
{
'companyName': 'MidSize Solutions',
'contactName': 'Emily Rodriguez',
'email': 'emily@midsize.com',
'title': 'VP of Marketing'
}
]
print(f"Importing {len(leads)} leads to campaign...")
results = client.add_leads_bulk(
os.getenv('MAGNETITE_CAMPAIGN_ID'),
leads,
delay=1.0 # 1 second between requests
)
# Summary
successful = [r for r in results if r['success']]
failed = [r for r in results if not r['success']]
print("\n=== IMPORT SUMMARY ===")
print(f"Total processed: {len(results)}")
print(f"Successful: {len(successful)}")
print(f"Failed: {len(failed)}")
if failed:
print("\nFailed leads:")
for f in failed:
print(f" - {f['lead']['email']}: {f['error']}")
return results
if __name__ == "__main__":
bulk_import()
print("\nBulk import complete!")CSV Import
Import leads from a CSV file
# csv_import.py
import os
import csv
from magnetite_client import MagnetiteClient
def import_from_csv(csv_path: str):
client = MagnetiteClient()
campaign_id = os.getenv('MAGNETITE_CAMPAIGN_ID')
leads = []
# Read CSV file
with open(csv_path, 'r') as file:
reader = csv.DictReader(file)
for row in reader:
lead = {
'companyName': row.get('company_name') or row.get('company'),
'contactName': row.get('contact_name') or row.get('name'),
'email': row.get('email'),
}
# Add optional fields if present
if row.get('title'):
lead['title'] = row['title']
if row.get('domain'):
lead['domain'] = row['domain']
if row.get('linkedin_url'):
lead['linkedInUrl'] = row['linkedin_url']
if row.get('notes'):
lead['notes'] = row['notes']
leads.append(lead)
print(f"Loaded {len(leads)} leads from CSV")
print(f"Importing to campaign {campaign_id}...")
results = client.add_leads_bulk(campaign_id, leads, delay=1.0)
# Summary
successful = [r for r in results if r['success']]
failed = [r for r in results if not r['success']]
print(f"\nImported {len(successful)} leads successfully")
print(f"Failed: {len(failed)}")
# Save failed leads for retry
if failed:
with open('failed_leads.csv', 'w', newline='') as f:
writer = csv.DictWriter(f, fieldnames=failed[0]['lead'].keys())
writer.writeheader()
for item in failed:
writer.writerow(item['lead'])
print("Failed leads saved to failed_leads.csv")
return results
if __name__ == "__main__":
import_from_csv('leads.csv')Flask Webhook Receiver
Handle meeting booked webhooks from Magnetite
# webhook_server.py
from flask import Flask, request, jsonify
import json
app = Flask(__name__)
@app.route('/webhooks/magnetite', methods=['POST'])
def handle_webhook():
event = request.json
print(f"Received webhook: {event.get('event')}")
if event.get('event') == 'meeting_booked':
prospect = event.get('prospect', {})
meeting = event.get('meeting', {})
campaign_id = event.get('campaignId')
print("Meeting booked!")
print(f"Company: {prospect.get('companyName')}")
print(f"Contact: {prospect.get('contactName')}")
print(f"Email: {prospect.get('email')}")
print(f"Scheduled: {meeting.get('scheduledAt')}")
print(f"Duration: {meeting.get('duration')} minutes")
# Handle the meeting booking
# - Update your CRM
# - Send notifications
# - Create calendar events
# etc.
# Example: send_slack_notification(prospect, meeting)
return jsonify({'received': True})
if __name__ == "__main__":
app.run(port=3000, debug=True)Next Steps
Explore more examples in different programming languages or dive deeper into the API.