An endpoint is an internet-accessible address, usually a URL, where an API receives a request to provide data or perform an action. It gives other applications a clear boundary for using your application’s logic. section 2.2
A typical endpoint interaction follows these stages:
- The calling application sends a request to a specific endpoint. section 2.2
- The request includes parameters describing the required data or action, and may include an API key for authentication. section 2.2
- The endpoint processes the call and returns data or performs the requested operation. section 2.2
Exposing a GenAI flow
A Langflow application can expose its LLM-based logic as an API endpoint so another application can call the flow in the background. In a coded version, FastAPI wraps the generation logic in a POST /generate-email route that accepts a customer profile and returns JSON containing an email subject and body. section 2.2 section 7.3
"""
Install dependencies once (pinning versions to match Langflow):
pip install "langchain==0.2.7" "langchain-openai==0.1.14" \
"langchain-community==0.2.7" pypdf fastapi uvicorn python-dotenv
Before running the service:
- keep `Luminia Guidelines.pdf` next to this script
- make sure OPENAI_API_KEY is available in `.env`
Start the API locally:
uvicorn CH7_LuminiaDEM:app --reload
Send a test request:
curl -X POST http://localhost:8000/generate-email \
-H "Content-Type: application/json" \
-d '{ "profile": { "name": "SANTOSHI", "segment": "LIGHTS", "recent_purchase_days_ago": 30 } }'
"""
import os #A
import json #B
from typing import Any, Dict #C
from dotenv import load_dotenv
from fastapi import FastAPI #D
from pydantic import BaseModel #E
from langchain_openai import ChatOpenAI #F
from langchain_core.prompts import ChatPromptTemplate #G
from langchain_community.document_loaders import PyPDFLoader #H
# --- Configuration ---------------------------------------------------------
BASE_DIR = os.path.dirname(os.path.abspath(__file__)) #I
GUIDELINES_PATH = os.path.join(BASE_DIR, "Luminia Guidelines.pdf") #J
# os.environ["OPENAI_API_KEY"] = "REPLACE_WITH_YOUR_KEY"
load_dotenv() #K
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2) #L
# --- Load and prepare guidelines text -------------------------------------
loader = PyPDFLoader(GUIDELINES_PATH) #M
docs = loader.load()
guidelines_text = "\n\n".join(page.page_content for page in docs) #N
PROMPT_TEMPLATE = """
Act as a senior copywriter for Luminia.
Your task is to generate a personalized marketing email that includes:
- A short, engaging subject line
- A concise body tailored to the customer profile (under 120 words)
Follow the brand rules strictly and reference the customer data provided.
Just generate the subject and the body as a JSON object.
The names of the keys must be: 'subject' and 'body'.
-------------------------------------------------
### CUSTOMER GUIDELINES
{guidelines}
-------------------------------------------------
""".strip() #O
prompt = ChatPromptTemplate.from_messages(
[
("system", PROMPT_TEMPLATE),
("human", "{customer_json}"),
]
) #P
# --- Core helper: turn a profile into subject+body ------------------------
def generate_email_from_profile(profile: Dict[str, Any]) - > Dict[str, str]:
"""Call the LLM to generate a JSON email (subject+body) for one customer."""
customer_json = json.dumps(profile, ensure_ascii=False, indent=2) #R
messages = prompt.format_messages(
guidelines=guidelines_text,
customer_json=customer_json,
) #S
response = llm.invoke(messages) #T
text = response.content.strip() #U
# We expect a JSON object like {"subject": "...", "body": "..."}.
try:
data = json.loads(text)
except json.JSONDecodeError:
# If parsing fails, return everything in the body and a generic subject.
return {
"subject": "Luminia special offer",
"body": text,
}
subject = data.get("subject", "Luminia special offer")
body = data.get("body", text)
return {"subject": subject, "body": body} #V
fallback.
# --- FastAPI app and request model ----------------------------------------
class CustomerProfile(BaseModel):
"""A generic customer profile wrapper."""
profile: Dict[str, Any] #W
app = FastAPI(title="Luminia Email Generator") #X
@app.post("/generate-email")
async def generate_email(payload: CustomerProfile) - > Dict[str, str]:
"""
Accepts a JSON customer profile and returns a brand-compliant
Luminia email with 'subject' and 'body' keys.
"""
return generate_email_from_profile(payload.profile) #Y
# --- Optional: quick local test without HTTP ------------------------------
if __name__ == "__main__": #Z
# Simple CLI test: type a minimal profile and see the result.
sample_profile = {
"name": "Alex",
"segment": "loyal_customer",
"last_purchase_days_ago": 45,
"preferred_channel": "email",
"interests": ["smart lighting", "energy savings"],
}
print("Sample profile:", json.dumps(sample_profile, indent=2))
email = generate_email_from_profile(sample_profile)
print("\nGenerated email JSON:\n", json.dumps(email, indent=2, ensure_ascii=False))
print(
"\nTo start the API server, run from the terminal:\n"
" uvicorn CH7_Luminia_Email_API:app --reload\n"
"Then POST to http://localhost:8000/generate-email with a JSON body like:\n"
'{ "profile": { ... your customer data ... } }'
)
Use the runnable listing to start the local API with Uvicorn and test the POST endpoint with curl. section 7.3
