FastAPI Integration Guide
Learn how to integrate Euler with your FastAPI backend to power the agentic chat functionality.
Illustrative example. The code below is a reference for building your own FastAPI backend — it is not a deployable service, and some handlers are stubs. Review and secure it before any production use.
Overview
Euler's backend is powered by FastAPI, a modern, fast web framework for building APIs with Python. This guide will help you set up a FastAPI backend that integrates with Euler's frontend components.
Prerequisites
- Python 3.8 or higher
- FastAPI
- Uvicorn (ASGI server)
- Supabase Python client
- An LLM provider (OpenAI, Anthropic, etc.)
You can install the required packages using pip:
pip install fastapi uvicorn supabase python-dotenv openai
Project Structure
Here's a recommended project structure for your FastAPI backend:
euler-fastapi/ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── dependencies.py │ ├── routers/ │ │ ├── __init__.py │ │ ├── chat.py │ │ ├── search.py │ │ ├── compare.py │ │ └── cart.py │ ├── models/ │ │ ├── __init__.py │ │ ├── chat.py │ │ ├── product.py │ │ └── cart.py │ ├── services/ │ │ ├── __init__.py │ │ ├── llm.py │ │ ├── search.py │ │ └── database.py │ └── utils/ │ ├── __init__.py │ └── helpers.py ├── .env ├── requirements.txt └── README.md
Setting Up the FastAPI App
First, let's set up the main FastAPI application:
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import chat, search, compare, cart
app = FastAPI(
title="Euler API",
description="API for Euler AI-powered e-commerce assistant",
version="1.0.0",
)
# Configure CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["https://www.eulerai.app"], # set to your frontend origin(s)
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers
app.include_router(chat.router, prefix="/api", tags=["chat"])
app.include_router(search.router, prefix="/api", tags=["search"])
app.include_router(compare.router, prefix="/api", tags=["compare"])
app.include_router(cart.router, prefix="/api", tags=["cart"])
@app.get("/")
async def root():
return {"message": "Welcome to Euler API"}
if __name__ == "__main__":
import uvicorn
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)Configuration
Create a configuration file to manage environment variables:
# app/config.py
import os
from dotenv import load_dotenv
load_dotenv()
class Settings:
PROJECT_NAME: str = "Euler API"
PROJECT_VERSION: str = "1.0.0"
SUPABASE_URL: str = os.getenv("SUPABASE_URL")
SUPABASE_KEY: str = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY")
# Add other configuration variables as needed
settings = Settings()Database Service
Create a service to interact with Supabase:
# app/services/database.py
from supabase import create_client
from app.config import settings
class DatabaseService:
def __init__(self):
self.supabase = create_client(settings.SUPABASE_URL, settings.SUPABASE_KEY)
async def get_products(self, filters=None, page=1, limit=10):
query = self.supabase.table("products").select("*")
if filters:
if "price" in filters:
if "min" in filters["price"]:
query = query.gte("price", filters["price"]["min"])
if "max" in filters["price"]:
query = query.lte("price", filters["price"]["max"])
if "categories" in filters:
query = query.in_("category", filters["categories"])
# Add more filters as needed
# Add pagination
start = (page - 1) * limit
query = query.range(start, start + limit - 1)
response = query.execute()
return response.data
async def get_product(self, product_id):
response = self.supabase.table("products").select("*").eq("id", product_id).execute()
if response.data:
return response.data[0]
return None
async def add_to_cart(self, user_id, product_id, quantity=1):
# Check if user has a cart
cart_response = self.supabase.table("carts").select("id").eq("user_id", user_id).execute()
if not cart_response.data:
# Create a new cart
cart_response = self.supabase.table("carts").insert({"user_id": user_id}).execute()
cart_id = cart_response.data[0]["id"]
# Check if product is already in cart
item_response = self.supabase.table("cart_items").select("*").eq("cart_id", cart_id).eq("product_id", product_id).execute()
if item_response.data:
# Update quantity
item_id = item_response.data[0]["id"]
new_quantity = item_response.data[0]["quantity"] + quantity
self.supabase.table("cart_items").update({"quantity": new_quantity}).eq("id", item_id).execute()
else:
# Add new item
self.supabase.table("cart_items").insert({
"cart_id": cart_id,
"product_id": product_id,
"quantity": quantity
}).execute()
# Return updated cart
return await self.get_cart(user_id)
async def get_cart(self, user_id):
# Get cart
cart_response = self.supabase.table("carts").select("id").eq("user_id", user_id).execute()
if not cart_response.data:
return {"items": [], "total_items": 0, "subtotal": 0}
cart_id = cart_response.data[0]["id"]
# Get cart items with product details
items_response = self.supabase.table("cart_items").select(
"id, quantity, products(id, name, price, image_url)"
).eq("cart_id", cart_id).execute()
items = []
subtotal = 0
for item in items_response.data:
product = item["products"]
quantity = item["quantity"]
total = product["price"] * quantity
subtotal += total
items.append({
"id": item["id"],
"product_id": product["id"],
"name": product["name"],
"price": product["price"],
"image_url": product["image_url"],
"quantity": quantity,
"total": total
})
return {
"cart_id": cart_id,
"items": items,
"total_items": len(items),
"subtotal": subtotal
}
db_service = DatabaseService()LLM Service
Create a service to interact with your LLM provider:
# app/services/llm.py
import openai
from app.config import settings
openai.api_key = settings.OPENAI_API_KEY
class LLMService:
async def generate_response(self, message, conversation_history=None):
"""
Generate a response using the LLM.
Args:
message: The user's message
conversation_history: Optional list of previous messages
Returns:
The LLM's response
"""
messages = []
# Add system message
messages.append({
"role": "system",
"content": "You are Euler, an AI shopping assistant. You help users find products, compare options, and make purchasing decisions. Be helpful, concise, and friendly."
})
# Add conversation history
if conversation_history:
for msg in conversation_history:
messages.append({
"role": msg["role"],
"content": msg["content"]
})
# Add user message
messages.append({
"role": "user",
"content": message
})
# Call OpenAI API
response = openai.ChatCompletion.create(
model="gpt-4", # or your preferred model
messages=messages,
max_tokens=500,
temperature=0.7
)
return {
"id": response.id,
"role": "assistant",
"content": response.choices[0].message.content,
"timestamp": response.created
}
llm_service = LLMService()Chat Router
Create a router for the chat functionality:
# app/routers/chat.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from app.services.llm import llm_service
from app.services.database import db_service
router = APIRouter()
class Message(BaseModel):
message: str
conversation_id: Optional[str] = None
user_id: Optional[str] = None
@router.post("/chat")
async def chat(message: Message):
"""
Process a chat message and return a response.
"""
try:
# Get conversation history if conversation_id is provided
conversation_history = []
if message.conversation_id and message.user_id:
# TODO: Fetch conversation history from database
pass
# Generate response
response = await llm_service.generate_response(message.message, conversation_history)
# Store message and response if user_id and conversation_id are provided
if message.user_id and message.conversation_id:
# TODO: Store message and response in database
pass
return response
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Search Router
Create a router for the search functionality:
# app/routers/search.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import List, Optional, Dict, Any
from app.services.database import db_service
router = APIRouter()
class SearchQuery(BaseModel):
query: str
filters: Optional[Dict[str, Any]] = None
page: Optional[int] = 1
limit: Optional[int] = 10
@router.post("/search")
async def search(query: SearchQuery):
"""
Search for products based on a query and filters.
"""
try:
# Process natural language query
# TODO: Use embeddings or NLP to extract search parameters from query
# For now, just pass filters directly
products = await db_service.get_products(query.filters, query.page, query.limit)
# Get total count for pagination
# TODO: Get total count from database
total = len(products)
return {
"products": products,
"total": total,
"page": query.page,
"limit": query.limit
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Compare Router
Create a router for the compare functionality:
# app/routers/compare.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import List
from app.services.database import db_service
router = APIRouter()
class CompareRequest(BaseModel):
product_ids: List[str]
@router.post("/compare")
async def compare(request: CompareRequest):
"""
Compare multiple products.
"""
try:
if len(request.product_ids) < 2:
raise HTTPException(status_code=400, detail="At least two products are required for comparison")
products = []
for product_id in request.product_ids:
product = await db_service.get_product(product_id)
if product:
products.append(product)
if len(products) < 2:
raise HTTPException(status_code=404, detail="Not all products were found")
# Generate comparison
comparison = {}
# Compare price
price_comparison = {}
for product in products:
price_comparison[product["id"]] = product["price"]
# Calculate price difference
prices = [product["price"] for product in products]
max_price = max(prices)
min_price = min(prices)
price_difference = max_price - min_price
percentage_difference = (price_difference / min_price) * 100 if min_price > 0 else 0
comparison["price"] = {
**price_comparison,
"difference": price_difference,
"percentage_difference": percentage_difference
}
# Compare attributes
attributes_comparison = {}
all_attributes = set()
for product in products:
if "attributes" in product and product["attributes"]:
for key in product["attributes"].keys():
all_attributes.add(key)
for attribute in all_attributes:
attribute_comparison = {}
for product in products:
if "attributes" in product and product["attributes"] and attribute in product["attributes"]:
attribute_comparison[product["id"]] = product["attributes"][attribute]
else:
attribute_comparison[product["id"]] = None
attributes_comparison[attribute] = attribute_comparison
comparison["attributes"] = attributes_comparison
# Compare ratings
if all("rating" in product for product in products):
rating_comparison = {}
for product in products:
rating_comparison[product["id"]] = product["rating"]
ratings = [product["rating"] for product in products]
max_rating = max(ratings)
min_rating = min(ratings)
rating_difference = max_rating - min_rating
comparison["rating"] = {
**rating_comparison,
"difference": rating_difference
}
return {
"products": products,
"comparison": comparison
}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Cart Router
Create a router for the cart functionality:
# app/routers/cart.py
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from typing import Optional
from app.services.database import db_service
router = APIRouter()
class CartItem(BaseModel):
user_id: str
product_id: str
quantity: Optional[int] = 1
@router.post("/cart/add")
async def add_to_cart(item: CartItem):
"""
Add a product to the cart.
"""
try:
cart = await db_service.add_to_cart(item.user_id, item.product_id, item.quantity)
return cart
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.get("/cart/{user_id}")
async def get_cart(user_id: str):
"""
Get the cart for a user.
"""
try:
cart = await db_service.get_cart(user_id)
return cart
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@router.delete("/cart/{user_id}/items/{item_id}")
async def remove_from_cart(user_id: str, item_id: str):
"""
Remove an item from the cart.
"""
try:
# TODO: Implement remove from cart functionality
return {"message": "Item removed from cart"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))Running the FastAPI Server
To run the FastAPI server, use the following command:
uvicorn app.main:app --reload
This will start the server at http://localhost:8000. You can access the API documentation at http://localhost:8000/docs.
Integrating with the Frontend
To integrate the FastAPI backend with the Euler frontend, you'll need to update the API endpoints in the frontend code. Here's an example of how to update the chat component:
// In your chat component
const handleSendMessage = async () => {
if (!input.trim() || isLoading) return;
// Add user message to chat
const userMessage = {
id: Date.now(),
role: "user",
content: input,
timestamp: new Date(),
};
setMessages((prev) => [...prev, userMessage]);
setInput("");
setIsLoading(true);
try {
// Call FastAPI endpoint
const response = await fetch("http://localhost:8000/api/chat", {
method: "POST",
headers: {
"Content-Type": "application/json",
},
body: JSON.stringify({
message: input,
conversation_id: conversationId,
user_id: userId,
}),
});
if (!response.ok) {
throw new Error("Failed to send message");
}
const data = await response.json();
setMessages((prev) => [...prev, data]);
} catch (error) {
console.error("Error sending message:", error);
// Add error message
setMessages((prev) => [
...prev,
{
id: Date.now(),
role: "assistant",
content: "Sorry, I encountered an error. Please try again later.",
timestamp: new Date(),
},
]);
} finally {
setIsLoading(false);
}
};Deployment
To deploy your FastAPI backend, you can use a variety of hosting options:
- Vercel: You can deploy FastAPI applications on Vercel using serverless functions.
- Heroku: Heroku provides a simple way to deploy FastAPI applications.
- AWS Lambda: You can deploy FastAPI applications as AWS Lambda functions using Mangum.
- Docker: You can containerize your FastAPI application and deploy it on any container-based hosting platform.
For production deployments, make sure to:
- Set appropriate CORS settings to allow only your frontend domain
- Use environment variables for sensitive information
- Implement proper authentication and authorization
- Set up monitoring and logging
Conclusion
This guide has shown you how to set up a FastAPI backend for Euler's AI-powered e-commerce assistant. By following these steps, you can create a robust backend that integrates with the Euler frontend and provides the necessary functionality for the agentic chat experience.
For more information, check out the API documentation or contact us for support.