CrewAI Cold Email: Build a Sales Crew with FoxReach

pip install crewai crewai-tools foxreach

import os
from crewai.tools import tool
from foxreach import FoxReach

client = FoxReach(api_key=os.environ["FOXREACH_API_KEY"])

@tool("Create a draft cold email campaign")
def create_campaign(name: str) -> str:
 """Create a new draft campaign in FoxReach. Returns the campaign ID."""
 campaign = client.campaigns.create(name=name, status="draft")
 return f"campaign_id={campaign.id}"

@tool("Add a sequence step to a campaign")
def add_sequence_step(campaign_id: str, subject: str, body: str, wait_days: int = 0) -> str:
 """Add an email step to a campaign sequence."""
 step = client.sequences.create(
 campaign_id=campaign_id, subject=subject, body=body, wait_days=wait_days,
 )
 return f"step_id={step.id} subject={step.subject!r}"

@tool("Import a lead into a campaign")
def import_lead(campaign_id: str, email: str, first_name: str, company: str) -> str:
 """Add a single lead to a campaign."""
 lead = client.leads.create(
 campaign_id=campaign_id, email=email, first_name=first_name, company=company,
 )
 return f"lead_id={lead.id}"

from crewai import Agent
from crewai_tools import SerperDevTool

search = SerperDevTool()

researcher = Agent(
 role="Outbound Researcher",
 goal="Find the single most specific and recent fact about a lead that justifies a personalized opener.",
 backstory="You read LinkedIn, Crunchbase, and TechCrunch daily. You know what makes a cold open land - a real project, a real promotion, a real funding round. You never invent. If the research is thin, you say so.",
 tools=[search],
 llm="gpt-4o-mini",
 verbose=True,
)

from crewai import Agent

copywriter = Agent(
 role="Cold Email Copywriter",
 goal="Write a 60-90 word cold email that feels written to one person, not blasted to a list.",
 backstory="You've written a million outbound emails and unsubscribed from every bad one. You open with one concrete observation, then a one-line value prop, then a soft ask. No 'hope this finds you well', no multi-paragraph intros, no questions that aren't answerable.",
 tools=[], # No tools - pure text in, text out.
 llm="gpt-4o",
 verbose=True,
)

from crewai import Agent
from foxreach_tools import create_campaign, add_sequence_step, import_lead

sender = Agent(
 role="Outbound Sender",
 goal="Create a FoxReach campaign in draft state with one personalized sequence step and one lead. Never call start_campaign - that's for humans.",
 backstory="You move drafts into FoxReach as drafts. You always name the campaign after the lead's company and the date. You never edit the copy the copywriter gave you.",
 tools=[create_campaign, add_sequence_step, import_lead],
 llm="gpt-4o-mini",
 verbose=True,
)

from crewai import Crew, Task, Process
from researcher import researcher
from copywriter import copywriter
from sender import sender

lead = {
 "name": "Alex Rivera",
 "title": "Head of Growth",
 "company": "Nuvoform",
 "email": "alex@nuvoform.ai",
}

research_task = Task(
 description=f"Research {lead['name']}, {lead['title']} at {lead['company']}. Return one specific, recent, citable fact about them or their company.",
 expected_output="A 40-80 word research note with one specific fact and its source.",
 agent=researcher,
)

copy_task = Task(
 description=f"""Using the research, write a cold email to {lead['name']} for Nuvoform. 60-90 words. Open with the research fact. Close with 'worth a quick chat?' or similar soft ask.""",
 expected_output="JSON with subject (max 60 chars) and body (60-90 words).",
 agent=copywriter,
 context=[research_task],
)

send_task = Task(
 description=f"Create a FoxReach campaign named 'Nuvoform - {lead['name']}'. Add the sequence step from the copywriter's output. Import lead {lead['email']} ({lead['name']}, {lead['company']}).",
 expected_output="campaign_id and lead_id from FoxReach.",
 agent=sender,
 context=[copy_task],
)

crew = Crew(
 agents=[researcher, copywriter, sender],
 tasks=[research_task, copy_task, send_task],
 process=Process.sequential,
 verbose=True,
)

result = crew.kickoff()
print(result)

import asyncio
from crewai import Crew, Task, Process

async def ship_one(lead: dict) -> dict:
 # Build per-lead tasks (same shape as above).
 crew = Crew(
 agents=[researcher, copywriter, sender],
 tasks=[...],
 process=Process.sequential,
 )
 return crew.kickoff() # CrewAI handles the concurrent LLM calls internally.

async def ship_many(leads: list[dict]) -> list[dict]:
 # Cap concurrency so we don't blow past FoxReach's 100 req/min rate limit.
 sem = asyncio.Semaphore(10)
 async def guarded(lead):
 async with sem:
 return await asyncio.to_thread(ship_one, lead)
 return await asyncio.gather(*(guarded(lead) for lead in leads))