OpenAI Agents SDK Cold Email: Production Outbound Agents with FoxReach

pip install openai-agents

import os
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

async def main():
 foxreach_mcp = MCPServerStreamableHttp(
 params={
 "url": "https://api.foxreach.io/mcp",
 "headers": {
 "Authorization": f"Bearer {os.environ['FOXREACH_API_KEY']}",
 },
 },
 name="foxreach",
 )

 async with foxreach_mcp:
 agent = Agent(
 name="Outbound Manager",
 instructions="You manage cold email campaigns. Use the foxreach MCP tools.",
 mcp_servers=[foxreach_mcp],
 model="gpt-4o-mini",
 )
 result = await Runner.run(
 agent,
 "Create a draft campaign 'Tier 2 Agencies' with a 3-step sequence. 3 days between each.",
 )
 print(result.final_output)

import asyncio
asyncio.run(main())

from pydantic import BaseModel
from agents import Agent

class ResearchResult(BaseModel):
 name: str
 company: str
 recent_signal: str
 source_url: str

researcher = Agent(
 name="Researcher",
 instructions="Find one specific, recent, citable fact about the lead. Return structured ResearchResult.",
 model="gpt-4o-mini",
 output_type=ResearchResult,
)

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

copywriter = Agent(
 name="Copywriter",
 instructions="Write a 70-word cold email using the research. Open with the research fact. Soft ask at the end.",
 model="gpt-4o",
 output_type=EmailDraft,
)

sender = Agent(
 name="Sender",
 instructions="Take the EmailDraft + lead info. Call foxreach MCP tools to create a draft campaign with one sequence step. Return the campaign_id.",
 mcp_servers=[foxreach_mcp],
 model="gpt-4o-mini",
 handoffs=[], # Terminal agent; no further handoffs.
)

# Orchestrator hands off sequentially:
orchestrator = Agent(
 name="Orchestrator",
 instructions="Research the lead, hand off to the copywriter, then hand off to the sender.",
 model="gpt-4o-mini",
 handoffs=[researcher, copywriter, sender],
)

result = await Runner.run(
 orchestrator,
 "Lead: Alex Rivera, Head of Growth at Nuvoform, alex@nuvoform.ai",
 max_turns=10,
)
print(result.final_output)

from agents import output_guardrail, GuardrailFunctionOutput

@output_guardrail
async def no_fabricated_numbers(ctx, agent, output: EmailDraft) -> GuardrailFunctionOutput:
 # Reject obvious statistics claims without citations.
 suspicious = ["%", "x increase", "million users", "faster than"]
 if any(s in output.body.lower() for s in suspicious):
 return GuardrailFunctionOutput(
 output_info={"reason": "unsourced claim detected"},
 tripwire_triggered=True,
 )
 return GuardrailFunctionOutput(output_info=None, tripwire_triggered=False)

copywriter = Agent(
 name="Copywriter",
 instructions="Write a 70-word cold email. No statistics, no percentages.",
 model="gpt-4o",
 output_type=EmailDraft,
 output_guardrails=[no_fabricated_numbers],
)