LangChain Cold Email: Build Production Outreach Agents with FoxReach

pip install foxreach langchain langchain-openai

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

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

@tool
def create_campaign(name: str) -> dict:
 """Create a new draft cold email campaign in FoxReach.

 The campaign starts in draft state. Call start_campaign to begin sending.
 """
 campaign = client.campaigns.create(name=name, status="draft")
 return {"id": campaign.id, "name": campaign.name, "status": campaign.status}

@tool
def add_sequence_step(
 campaign_id: str,
 subject: str,
 body: str,
 wait_days: int = 3,
) -> dict:
 """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 {"id": step.id, "subject": step.subject}

@tool
def import_leads(campaign_id: str, leads: list[dict]) -> dict:
 """Import leads into a campaign. Each lead needs at minimum an email field."""
 result = client.leads.create_batch(campaign_id=campaign_id, leads=leads)
 return {"imported": len(result.created), "skipped": len(result.skipped)}

pip install langchain-mcp-adapters langchain langchain-openai

import os
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

async def main():
 client = MultiServerMCPClient({
 "foxreach": {
 "url": "https://api.foxreach.io/mcp",
 "transport": "streamable_http",
 "headers": {
 "Authorization": f"Bearer {os.environ['FOXREACH_API_KEY']}",
 },
 },
 })
 tools = await client.get_tools() # returns all 23 FoxReach MCP tools

 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 prompt = ChatPromptTemplate.from_messages([
 ("system", "You are an outbound sales agent. Use FoxReach tools to manage campaigns and leads."),
 ("human", "{input}"),
 ("placeholder", "{agent_scratchpad}"),
 ])

 agent = create_tool_calling_agent(llm, tools, prompt)
 executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

 await executor.ainvoke({
 "input": "Create a draft campaign called 'SaaS CTOs April' and add a 3-step sequence: intro, value prop, soft ask. Three days between each.",
 })

if __name__ == "__main__":
 asyncio.run(main())

from langchain_core.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults

search = TavilySearchResults(max_results=3)

@tool
def research_lead(name: str, company: str) -> str:
 """Find recent public context about a person and their company."""
 query = f"{name} {company} site:linkedin.com OR site:crunchbase.com"
 results = search.invoke(query)
 return "\n\n".join(r["content"] for r in results)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel

class PersonalizedEmail(BaseModel):
 subject: str
 body: str

llm = ChatOpenAI(model="gpt-4o-mini").with_structured_output(PersonalizedEmail)
prompt = ChatPromptTemplate.from_messages([
 ("system", "You write concise, specific cold emails. Open with one concrete observation from the research. Under 90 words."),
 ("human", "Lead: {name}, {title} at {company}.\nResearch:\n{research}\n\nTemplate:\n{template}"),
])

@tool
def personalize_email(
 name: str, title: str, company: str, research: str, template: str,
) -> dict:
 """Rewrite a template into a personalized email for this lead."""
 chain = prompt | llm
 result = chain.invoke({"name": name, "title": title, "company": company, "research": research, "template": template})
 return {"subject": result.subject, "body": result.body}

from foxreach import FoxReach

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

def ship_personalized_campaign(lead: dict, template: str):
 research = research_lead.invoke({
 "name": lead["name"], "company": lead["company"],
 })
 email = personalize_email.invoke({
 "name": lead["name"], "title": lead["title"],
 "company": lead["company"], "research": research, "template": template,
 })

 campaign = client.campaigns.create(
 name=f"Personalized - {lead['company']}",
 status="draft",
 )
 client.sequences.create(
 campaign_id=campaign.id,
 subject=email["subject"],
 body=email["body"],
 wait_days=0,
 )
 client.leads.create(
 campaign_id=campaign.id,
 email=lead["email"],
 first_name=lead["name"],
 company=lead["company"],
 )
 return {"campaign_id": campaign.id, "status": "draft_ready_for_review"}