`
2. **REQUIRED PLACEHOLDERS:** You MUST include `{{Welcome_Message}}` and `{{Closing_Message}}` in the first email. You should also use `{{first_name}}` in follow-ups.
3. **FIRST EMAIL STRUCTURE:** The first email's body MUST begin with `{{Welcome_Message}}` and end with `{{Closing_Message}}`.
4. **SIGNATURE:** End EVERY email body with `
Best regards`. DO NOT WRITE NAME AND TITLE AS IT WILL BE ALREADY ADDED BY DEFAULT
5. **CONTENT REQUIREMENTS:** Each email must mention:
- Company name
- Role title
- Work location (onsite/hybrid/remote)
- Company mission (in emails 2 and 3)
- Salary information or Compensation (if available, in all the emails)
- His/Her responsibilities in the company
6. **CALL-TO-ACTION:** Each email must end with the CTA (before the signature): "If you're interested, send me your updated CV, salary expectations, and I'll get your application in front of the hiring manager ASAP."
7. **EXAMPLE BODY:**
```html
{{Welcome_Message}}
I saw your profile and was impressed. We have an opening for a Senior Engineer at [Company Name] that seems like a great fit. This is a [remote/hybrid/onsite] position.
If you're interested, send me your updated CV, salary expectations, and I'll get your application in front of the hiring manager ASAP.
{{Closing_Message}}
Best regards,
Ali Taghikhani
CEO, SRN
```
Always try to start the message with the salutation except for the first email. If lead context is provided, use it to make the templates more relevant. Respond with ONLY a valid JSON object matching the required structure.
"""
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "Generate the 3-step email sequence template for this job description: {job_description}{lead_context}")
])
# By using .partial, we tell LangChain to treat the Smartlead placeholders as literals
# and not expect them as input variables. This is the correct way to handle this.
partial_prompt = prompt.partial(
first_name="",
company_name="",
**{"Welcome_Message": "", "Closing_Message": "", "Title": ""}
)
chain = partial_prompt | structured_llm
response = await chain.ainvoke({"job_description": job_description, "lead_context": lead_context})
# Post-process the AI's response to enforce double curly braces.
# This is a robust way to fix the AI's tendency to use single braces.
def fix_braces(text: str) -> str:
if not text:
return ""
# This regex finds all occurrences of `{...}` that are not `{{...}}`
# and replaces them with `{{...}}`.
return re.sub(r'{([^{}\n]+)}', r'{{\1}}', text)
sequences = [
CampaignSequence(
seq_number=1,
seq_delay_details=SeqDelayDetails(delay_in_days=1),
seq_variants=[SeqVariant(
subject=fix_braces(response.introduction.subject),
email_body=fix_braces(response.introduction.body),
variant_label="A"
)]
),
CampaignSequence(
seq_number=2,
seq_delay_details=SeqDelayDetails(delay_in_days=3),
subject="", # Same thread
email_body=fix_braces(response.follow_up_1.body)
),
CampaignSequence(
seq_number=3,
seq_delay_details=SeqDelayDetails(delay_in_days=5),
subject="", # Same thread
email_body=fix_braces(response.follow_up_2.body)
)
]
return sequences
except Exception as e:
print(f"Error generating sequences with LLM: {str(e)}. Falling back to template.")
return generate_template_sequences(job_description)
def generate_template_sequences(job_description: str) -> List[CampaignSequence]:
"""Generate template-based sequences as fallback, using correct placeholders."""
# This is the corrected structure for the first email
first_email_body = f"""{{{{custom.Welcome_Message}}}}
I'm reaching out because we have some exciting opportunities for a {job_description} that might be a great fit for your background. Are you currently open to exploring new roles?
{{{{custom.Closing_Message}}}}
Best regards,
Ali Taghikhani
CEO, SRN
"""
follow_up_1_body = f"""Hi {{{{first_name}}}},
Just wanted to follow up on my previous email regarding the {job_description} role. I'd love to hear your thoughts when you have a moment.
Best regards,
Ali Taghikhani
CEO, SRN
"""
follow_up_2_body = f"""Hi {{{{first_name}}}},
Checking in one last time about the {job_description} opportunity. If the timing isn't right, no worries at all. Otherwise, I look forward to hearing from you.
Best regards,
Ali Taghikhani
CEO, SRN
"""
sequences = [
CampaignSequence(
seq_number=1,
seq_delay_details=SeqDelayDetails(delay_in_days=1),
seq_variants=[
SeqVariant(
subject=f"Regarding a {job_description} opportunity",
email_body=first_email_body,
variant_label="A"
)
]
),
CampaignSequence(
seq_number=2,
seq_delay_details=SeqDelayDetails(delay_in_days=3),
subject="",
email_body=follow_up_1_body
),
CampaignSequence(
seq_number=3,
seq_delay_details=SeqDelayDetails(delay_in_days=5),
subject="",
email_body=follow_up_2_body
)
]
return sequences
# ============================================================================
# RATE LIMITING MIDDLEWARE
# ============================================================================
class RateLimiter:
def __init__(self, max_requests: int = 10, window_seconds: int = 2):
self.max_requests = max_requests
self.window_seconds = window_seconds
self.requests = []
def is_allowed(self) -> bool:
now = time.time()
self.requests = [req_time for req_time in self.requests if now - req_time < self.window_seconds]
if len(self.requests) >= self.max_requests:
return False
self.requests.append(now)
return True
rate_limiter = RateLimiter(max_requests=10, window_seconds=2)
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
"""Rate limiting middleware to respect Smartlead's API limits"""
if not rate_limiter.is_allowed():
return JSONResponse(
status_code=429,
content={"error": "Rate limit exceeded"}
)
response = await call_next(request)
return response
# ============================================================================
# ERROR HANDLING
# ============================================================================
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
"""Custom HTTP exception handler"""
return JSONResponse(
status_code=exc.status_code,
content={"error": True, "message": exc.detail}
)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""General exception handler"""
return JSONResponse(
status_code=500,
content={
"error": True,
"message": "Internal server error",
"detail": str(exc) if os.getenv("DEBUG") else None
}
)
# ============================================================================
# CUSTOM OPENAPI SCHEMA
# ============================================================================
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Smartlead API - Complete Integration",
version="2.1.0",
description="A comprehensive FastAPI wrapper for the Smartlead email automation platform.",
routes=app.routes,
)
openapi_schema["tags"] = [
{"name": "Campaigns", "description": "Campaign management operations"},
{"name": "Leads", "description": "Lead management operations"},
{"name": "Sequences", "description": "Email sequence management"},
{"name": "Webhooks", "description": "Webhook management"},
{"name": "Clients", "description": "Client account management"},
{"name": "Messages", "description": "Message history and reply operations"},
{"name": "Analytics", "description": "Campaign analytics and statistics"},
{"name": "Email Accounts", "description": "Email account management"},
{"name": "Utilities", "description": "Utility endpoints"}
]
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
# ============================================================================
# MAIN APPLICATION ENTRY POINT
# ============================================================================
if __name__ == "__main__":
import uvicorn
print("Starting Smartlead API - Complete Integration")
uvicorn.run(
"__main__:app",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info"
)