LangGraph Cold Email: Multi-Step Outreach Agents with FoxReach

pip install langgraph langchain langchain-openai foxreach

from typing import TypedDict, Literal, Optional

ReplyIntent = Literal["hot", "cold", "unsubscribe", "ooo", "unknown"]

class CampaignState(TypedDict):
 lead_id: str
 lead_email: str
 lead_name: str
 lead_company: str
 research_notes: Optional[str]
 draft_subject: Optional[str]
 draft_body: Optional[str]
 foxreach_campaign_id: Optional[str]
 last_reply: Optional[str]
 reply_intent: Optional[ReplyIntent]
 retries: int
 status: Literal["researching", "drafting", "sent", "waiting", "replied", "done"]

from foxreach import FoxReach
from langchain_openai import ChatOpenAI
import os

client = FoxReach(api_key=os.environ["FOXREACH_API_KEY"])
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
copywriter_llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

def research_node(state: CampaignState) -> dict:
 # Call your research tool here (Tavily, Serper, etc.)
 notes = f"Found recent signal about {state['lead_company']}..."
 return {"research_notes": notes, "status": "drafting"}

def draft_node(state: CampaignState) -> dict:
 prompt = f"Write a 70-word cold email to {state['lead_name']} at {state['lead_company']}. Research: {state['research_notes']}"
 result = copywriter_llm.invoke(prompt)
 # Assume result.content is "Subject: ...\n\nBody:..." - parse accordingly
 subject, body = result.content.split("\n\n", 1)
 return {"draft_subject": subject.replace("Subject: ", ""), "draft_body": body}

def send_node(state: CampaignState) -> dict:
 campaign = client.campaigns.create(
 name=f"{state['lead_company']} - Auto",
 status="draft",
 )
 client.sequences.create(
 campaign_id=campaign.id,
 subject=state["draft_subject"],
 body=state["draft_body"],
 )
 client.leads.create(
 campaign_id=campaign.id,
 email=state["lead_email"],
 first_name=state["lead_name"],
 company=state["lead_company"],
 )
 client.campaigns.start(campaign.id)
 return {"foxreach_campaign_id": campaign.id, "status": "waiting"}

def classify_reply_node(state: CampaignState) -> dict:
 prompt = f"Classify this cold email reply as hot, cold, unsubscribe, ooo, or unknown. Reply:\n{state['last_reply']}"
 intent = llm.invoke(prompt).content.strip().lower()
 return {"reply_intent": intent, "status": "replied"}

def route_reply(state: CampaignState) -> str:
 if state["reply_intent"] == "hot":
 return "draft_reply"
 if state["reply_intent"] == "unsubscribe":
 return "done"
 if state["retries"] >= 3:
 return "done"
 return "draft_follow_up"

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(CampaignState)

workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.add_node("send", send_node)
workflow.add_node("classify_reply", classify_reply_node)
workflow.add_node("draft_reply", draft_reply_node) # hot reply handler
workflow.add_node("draft_follow_up", draft_follow_up_node) # cold reply handler

workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", "send")
# send → wait state handled externally by the webhook receiver
# When the webhook fires, it resumes the graph and routes to classify_reply.
workflow.add_conditional_edges(
 "classify_reply",
 route_reply,
 {
 "draft_reply": "draft_reply",
 "draft_follow_up": "draft_follow_up",
 "done": END,
 },
)
workflow.add_edge("draft_reply", "send")
workflow.add_edge("draft_follow_up", "send")

graph = workflow.compile(checkpointer=MemorySaver())

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/foxreach/webhook")
async def foxreach_webhook(req: Request):
 payload = await req.json()
 if payload["event"] != "reply_received":
 return {"ok": True}
 lead_id = payload["lead_id"]
 reply_text = payload["message"]["body"]

 # Resume the graph for this lead. The thread_id binds state to the lead.
 result = graph.invoke(
 {"last_reply": reply_text},
 config={"configurable": {"thread_id": lead_id}},
 )
 return {"ok": True, "next_status": result["status"]}

graph = workflow.compile(
 checkpointer=MemorySaver(),
 interrupt_before=["send"], # Pause before send runs.
)

# Run until the interrupt:
result = graph.invoke(initial_state, config={"configurable": {"thread_id": lead_id}})
# result contains the drafted subject + body. Post it to Slack for approval.

# When the human approves:
graph.invoke(None, config={"configurable": {"thread_id": lead_id}})
# LangGraph resumes from the send node and continues.