catalyst-sdk 0.1.0

AI-powered outbound prospect finder - Generate Google dorks, find emails, and craft personalized outreach

Catalyst SDK (Python)

AI-powered outbound prospect finder. Generate Google dorks, find emails, and craft hyper-personalized outreach messages that don't sound like AI.

Installation

pip install catalyst-sdk
# or
poetry add catalyst-sdk
# or
uv add catalyst-sdk

---

🔑 API Keys Setup

Catalyst uses multiple AI and data services. Here's how to get each API key:

1. Google Gemini (Required)

Gemini AI powers the intelligent search strategy, dork generation, and message personalization.

1. Go to Google AI Studio
2. Click "Create API Key"
3. Copy your key

Model Used: gemma-3-27b-it (configurable)

2. SerpAPI (Optional - for Google searches)

SerpAPI executes Google searches to find prospect profiles.

1. Go to serpapi.com
2. Sign up for a free account (100 searches/month free)
3. Go to Dashboard → API Key
4. Copy your key

Without SerpAPI: You can still generate dorks and use them manually in Google.

3. Hunter.io (Optional - for email finding)

Hunter.io finds verified email addresses for your prospects.

1. Go to hunter.io
2. Sign up for a free account (25 searches/month free)
3. Go to API → Your API Key
4. Copy your key

Without Hunter: You'll need to find emails manually or use other email finding tools.

---

🚀 Quick Start

import asyncio
from catalyst import Catalyst

async def main():
    # Initialize with your API keys
    catalyst = Catalyst(
        gemini_api_key="your-gemini-key",      # Required
        serpapi_key="your-serpapi-key",         # Optional
        hunter_api_key="your-hunter-key",       # Optional
    )
    
    # Find prospects
    prospects = await catalyst.find_prospects("software engineers at Stripe", count=5)
    
    print(prospects)
    # [
    #   SocialProfile(
    #     platform='linkedin',
    #     profile_url='https://linkedin.com/in/johndoe',
    #     username='johndoe',
    #     full_name='John Doe',
    #     snippet='Senior Engineer at Stripe. Building payments infrastructure...'
    #   ),
    #   ...
    # ]
    
    # Generate personalized outreach
    message = await catalyst.generate_message(
        about_me="I'm building dev tools for auth. Previously at AWS.",
        prospect=prospects[0]
    )
    
    print(message)
    # "your work on stripe's payment infra is solid - we're tackling auth challenges 
    #  and would love to pick your brain. 15 min call?"

asyncio.run(main())

---

📖 Full Usage Guide

Option 1: Class-based API (Recommended)

import asyncio
from catalyst import Catalyst

async def main():
    catalyst = Catalyst(
        gemini_api_key="your-gemini-key",
        serpapi_key="your-serpapi-key",        # optional
        hunter_api_key="your-hunter-key",      # optional
        model="gemma-3-27b-it",                # optional, this is the default
    )
    
    # Full pipeline: strategy → dorks → search → extract
    prospects = await catalyst.find_prospects("YC founders in fintech", count=10)
    
    # Generate outreach message
    message = await catalyst.generate_message("About me text...", prospects[0])
    
    # Generate cold email
    email = await catalyst.generate_email(
        about_me="About me text...",
        prospect=prospects[0],
        email="[e-mail]"
    )
    # Email(subject='...', body='...')
    
    # Find email using Hunter.io
    email_result = await catalyst.find_email("stripe.com", "John", "Doe")
    # HunterResult(email='[e-mail]', score=95, ...)

asyncio.run(main())

Option 2: Functional API

For more control over individual steps:

import asyncio
import os
from catalyst import init_catalyst
from catalyst.services import (
    generate_search_strategy,
    generate_dorks,
    execute_search,
    extract_profiles,
    generate_personalized_message,
    generate_email,
    find_email,
)
from catalyst.services.twitter import (
    split_name,
    extract_company_domain,
)

async def main():
    # Initialize once at app startup
    init_catalyst(
        gemini_api_key=os.environ["GEMINI_API_KEY"],
        serpapi_key=os.environ.get("SERPAPI_KEY"),
        hunter_api_key=os.environ.get("HUNTER_API_KEY"),
    )
    
    # Step 1: Get AI-recommended platforms
    strategy = await generate_search_strategy("ML engineers at Google")
    print(strategy)
    # SearchStrategy(
    #   platforms=['linkedin'], 
    #   reasoning='LinkedIn is best for professional/employment outreach'
    # )
    
    # Step 2: Generate Google dorks
    dorks = await generate_dorks("ML engineers at Google", strategy.platforms)
    print(dorks.linkedin)
    # 'site:linkedin.com/in/ "machine learning" "google" "engineer"'
    
    # Step 3: Execute search (requires SerpAPI key)
    search_results = await execute_search(dorks.linkedin, num_results=20)
    # [SearchResult(title=..., link=..., snippet=..., position=...), ...]
    
    # Step 4: Extract profiles from results
    profiles = await extract_profiles(search_results, "linkedin", target_count=5)
    # [SocialProfile(...), ...]
    
    # Step 5: Find email (requires Hunter API key)
    first_name, last_name = split_name(profiles[0].full_name)
    domain = await extract_company_domain(profiles[0].snippet)
    if domain:
        email_result = await find_email(domain, first_name, last_name)
        print(email_result.email)  # '[e-mail]'
    
    # Step 6: Generate personalized message
    message = await generate_personalized_message(
        about_me="I run an ML infra startup. Ex-Meta AI team.",
        prospect=profiles[0]
    )
    
    # Step 7: Generate email
    email = await generate_email(
        about_me="I run an ML infra startup. Ex-Meta AI team.",
        prospect=profiles[0],
        email="[e-mail]"
    )

asyncio.run(main())

---

🔧 API Reference

init_catalyst(config) / Catalyst(config)

Initialize the SDK with your API keys.

from catalyst import Catalyst, init_catalyst

# Class-based
catalyst = Catalyst(
    gemini_api_key: str,       # Required - Google Gemini API key
    serpapi_key: str = None,   # Optional - SerpAPI key for searches
    hunter_api_key: str = None,# Optional - Hunter.io key for emails
    model: str = "gemma-3-27b-it",  # Optional - Gemini model
)

# Functional
init_catalyst(
    gemini_api_key: str,
    serpapi_key: str = None,
    hunter_api_key: str = None,
    model: str = "gemma-3-27b-it",
)

generate_search_strategy(query) -> SearchStrategy

Uses AI to determine the best platform(s) for your search query.

strategy = await generate_search_strategy("senior React developers")
# SearchStrategy(platforms=['linkedin'], reasoning='LinkedIn best for professional hiring')

Platform Selection Logic:

Target	Recommended Platform
Developers/Engineers	LinkedIn
Open Source Contributors	GitHub
Designers	Dribbble, Behance
Founders/VCs	Twitter/X
Enterprise/Sales	LinkedIn
Creators/Influencers	Instagram, Threads
Writers/Bloggers	Medium
Product People	Product Hunt

generate_dorks(query, platforms) -> DorkResult

Generate optimized Google dork queries.

dorks = await generate_dorks("ML engineers at Google", ["linkedin", "github"])
# DorkResult(
#   linkedin='site:linkedin.com/in/ "machine learning" "google"',
#   github='site:github.com "google" "ML" "tensorflow"'
# )

execute_search(query, num_results) -> List[SearchResult]

Execute a Google search via SerpAPI. Requires SerpAPI key.

results = await execute_search('site:linkedin.com/in/ "stripe"', num_results=10)
# [SearchResult(title=..., link=..., snippet=..., position=...), ...]

extract_profiles(serp_results, platform, target_count) -> List[SocialProfile]

Extract structured profile data from search results using AI.

profiles = await extract_profiles(results, "linkedin", target_count=5)
# [SocialProfile(
#     platform='linkedin',
#     profile_url='https://linkedin.com/in/johndoe',
#     username='johndoe', 
#     full_name='John Doe',
#     snippet='Senior Engineer at Stripe...'
# ), ...]

generate_personalized_message(about_me, prospect) -> str

Generate a hyper-personalized DM/message that doesn't sound like AI.

message = await generate_personalized_message(
    about_me="Building AI tools for sales. Ex-Salesforce.",
    prospect=SocialProfile(full_name="John Doe", platform="twitter", snippet="ML engineer @OpenAI")
)
# "saw your work on GPT-4 - we're building AI for sales, would love your take. quick call?"

generate_email(about_me, prospect, email) -> Email

Generate a personalized cold email with subject and body.

email = await generate_email(
    about_me="Founder at Acme. Building dev tools.",
    prospect=SocialProfile(full_name="Jane Smith", snippet="Auth lead at Stripe"),
    email="[e-mail]"
)
# Email(subject='your auth work + our sdk', body='...')

find_email(domain, first_name, last_name) -> HunterResult

Find someone's email using Hunter.io. Requires Hunter API key.

result = await find_email("stripe.com", "John", "Doe")
# HunterResult(email='[e-mail]', score=95, domain='stripe.com', ...)

extract_company_domain(snippet) -> str

Use AI to extract company domain from a profile snippet.

domain = await extract_company_domain("Senior Engineer at Stripe. Building payments...")
# 'stripe.com'

split_name(full_name) -> Tuple[str, str]

Split a full name into first and last name.

first_name, last_name = split_name("John Michael Doe")
# ('John', 'Michael Doe')

---

🐦 Twitter Utilities

from catalyst.services.twitter import (
    get_twitter_numeric_id,
    generate_twitter_dm_link,
    generate_twitter_profile_link,
    extract_twitter_username,
)

# Get numeric ID (required for DM deep links)
numeric_id = await get_twitter_numeric_id("elonmusk")
# '44196397'

# Generate DM deep link
dm_link = generate_twitter_dm_link(numeric_id, "Hey! Quick question...")
# 'https://x.com/messages/compose?recipient_id=44196397&text=Hey!%20Quick%20question...'

# Generate profile link
profile_link = generate_twitter_profile_link("naval")
# 'https://x.com/naval'

# Extract username from URL
username = extract_twitter_username("https://x.com/naval")
# 'naval'

---

🌐 Supported Platforms

Platform	Dork Generation	Profile Extraction	DM Deep Link
LinkedIn	✅	✅	❌
Twitter/X	✅	✅	✅
GitHub	✅	✅	❌
Instagram	✅	✅	✅
Dribbble	✅	✅	❌
Behance	✅	✅	❌
Medium	✅	✅	❌
Product Hunt	✅	✅	❌
Peerlist	✅	✅	❌
Threads	✅	✅	❌
Bluesky	✅	✅	❌

---

💡 Example: Full Outbound Workflow

import asyncio
import os
from catalyst import Catalyst

async def run_outbound_campaign():
    catalyst = Catalyst(
        gemini_api_key=os.environ["GEMINI_API_KEY"],
        serpapi_key=os.environ["SERPAPI_KEY"],
        hunter_api_key=os.environ["HUNTER_API_KEY"],
    )
    
    # Your pitch
    about_me = """
        I'm the founder of AuthKit, we're building developer-friendly authentication.
        Previously led identity at AWS. YC W24.
    """
    
    # Find prospects
    prospects = await catalyst.find_prospects(
        "senior engineers who worked on authentication at Stripe or Okta",
        count=10
    )
    
    print(f"Found {len(prospects)} prospects")
    
    for prospect in prospects:
        # Try to find their email
        name_parts = prospect.full_name.split(" ", 1)
        first_name = name_parts[0]
        last_name = name_parts[1] if len(name_parts) > 1 else ""
        
        email_result = await catalyst.find_email("stripe.com", first_name, last_name)
        
        if email_result:
            # Generate personalized email
            outreach = await catalyst.generate_email(
                about_me=about_me,
                prospect=prospect,
                email=email_result.email
            )
            
            print(f"\n--- {prospect.full_name} ({email_result.email}) ---")
            print(f"Subject: {outreach.subject}")
            print(f"Body: {outreach.body}")
        else:
            # No email? Generate a DM instead
            message = await catalyst.generate_message(about_me, prospect)
            print(f"\n--- {prospect.full_name} (DM) ---")
            print(f"Message: {message}")

asyncio.run(run_outbound_campaign())

---

🔒 Environment Variables

Create a .env file:

# Required
GEMINI_API_KEY=your-gemini-api-key

# Optional (but recommended for full functionality)
SERPAPI_KEY=your-serpapi-key
HUNTER_API_KEY=your-hunter-api-key

Load with python-dotenv:

from dotenv import load_dotenv
import os
from catalyst import Catalyst

load_dotenv()

catalyst = Catalyst(
    gemini_api_key=os.environ["GEMINI_API_KEY"],
    serpapi_key=os.environ.get("SERPAPI_KEY"),
    hunter_api_key=os.environ.get("HUNTER_API_KEY"),
)

---

📊 Rate Limits & Pricing

Service	Free Tier	Paid Plans
Gemini	60 req/min	Pricing
SerpAPI	100 searches/mo	$50/mo for 5000
Hunter	25 searches/mo	$49/mo for 500

---

📦 Type Definitions

All types are defined using Pydantic for runtime validation:

from catalyst.types import (
    SocialProfile,
    SearchResult,
    SearchStrategy,
    DorkResult,
    HunterResult,
    Email,
)

---

⚠️ Important: Sending vs Generating

This SDK generates content but does NOT send emails or DMs.

What Catalyst Does	What You Need to Add
✅ Find prospects	❌ N/A
✅ Find emails (Hunter)	❌ N/A
✅ Generate personalized messages	📧 Your own email/DM sending
✅ Generate cold emails	📧 Your own email sending

Why?

1. OAuth requires a web app - Gmail/Twitter auth needs redirects and user consent
2. You control sending - Use your own domain, avoid spam filters
3. Legal compliance - CAN-SPAM, GDPR require sender accountability

Integrating Email Sending

Use the generated content with your preferred email service:

Option 1: smtplib (Built-in)

import smtplib
from email.mime.text import MIMEText
from catalyst import Catalyst

catalyst = Catalyst(gemini_api_key=os.environ["GEMINI_API_KEY"])

# Generate the email
email = await catalyst.generate_email(about_me, prospect, prospect_email)

# Send via SMTP
msg = MIMEText(email.body)
msg['Subject'] = email.subject
msg['From'] = '[e-mail]'
msg['To'] = prospect_email

with smtplib.SMTP('smtp.gmail.com', 587) as server:
    server.starttls()
    server.login('[e-mail]', os.environ['EMAIL_APP_PASSWORD'])
    server.send_message(msg)

Option 2: SendGrid

from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail
from catalyst import Catalyst

catalyst = Catalyst(gemini_api_key=os.environ["GEMINI_API_KEY"])
email = await catalyst.generate_email(about_me, prospect, prospect_email)

sg = SendGridAPIClient(os.environ['SENDGRID_API_KEY'])
message = Mail(
    from_email='[e-mail]',
    to_emails=prospect_email,
    subject=email.subject,
    plain_text_content=email.body
)
sg.send(message)

Option 3: Resend

import resend
from catalyst import Catalyst

resend.api_key = os.environ["RESEND_API_KEY"]

catalyst = Catalyst(gemini_api_key=os.environ["GEMINI_API_KEY"])
email = await catalyst.generate_email(about_me, prospect, prospect_email)

resend.Emails.send({
    "from": "[e-mail]",
    "to": prospect_email,
    "subject": email.subject,
    "text": email.body
})

Option 4: AWS SES (boto3)

import boto3
from catalyst import Catalyst

ses = boto3.client('ses', region_name='us-east-1')

catalyst = Catalyst(gemini_api_key=os.environ["GEMINI_API_KEY"])
email = await catalyst.generate_email(about_me, prospect, prospect_email)

ses.send_email(
    Source='[e-mail]',
    Destination={'ToAddresses': [prospect_email]},
    Message={
        'Subject': {'Data': email.subject},
        'Body': {'Text': {'Data': email.body}}
    }
)

Sending Twitter/LinkedIn DMs

For DMs, you'll need to either:

1. Use the generated message manually - Copy/paste into the platform
2. Use platform APIs - Requires OAuth setup in your own app
3. Use automation tools - Phantombuster, LinkedIn Sales Navigator, etc.

from catalyst import Catalyst
from catalyst.services.twitter import get_twitter_numeric_id, generate_twitter_dm_link

catalyst = Catalyst(gemini_api_key=os.environ["GEMINI_API_KEY"])

# Generate a DM for Twitter
message = await catalyst.generate_message(about_me, prospect)

# Get Twitter DM deep link (opens Twitter app/web with pre-filled message)
numeric_id = await get_twitter_numeric_id(prospect.username)
if numeric_id:
    dm_link = generate_twitter_dm_link(numeric_id, message)
    print(f'Open this to send DM: {dm_link}')
    # https://x.com/messages/compose?recipient_id=123&text=your%20message

---

🤝 Contributing

PRs welcome! Please read our contributing guidelines first.

📄 License

MIT © Namit