"""
FastMCP HTTP Server for Product Recommendation System
Provides tools for searching and filtering products using ChromaDB vector store
"""
import os
from typing import List, Dict, Optional
from fastmcp import FastMCP
import chromadb
from openai import AzureOpenAI
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
# Initialize FastMCP server
mcp = FastMCP("Product Recommendation Server")
# Global variables for ChromaDB and Azure OpenAI
chroma_client = None
collection = None
azure_client = None
embedding_deployment = None
top_k = None
def initialize_services():
"""Initialize ChromaDB and Azure OpenAI clients"""
global chroma_client, collection, azure_client, embedding_deployment, top_k
# Azure OpenAI configuration
azure_client = AzureOpenAI(
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT")
)
embedding_deployment = os.getenv("AZURE_OPENAI_EMBEDDING_DEPLOYMENT")
# ChromaDB configuration
persist_directory = os.getenv("CHROMA_PERSIST_DIRECTORY", "./chroma_db")
chroma_client = chromadb.PersistentClient(path=persist_directory)
collection = chroma_client.get_collection(name="products")
# Top K configuration
top_k = int(os.getenv("TOP_K", "50"))
print(f"✓ Services initialized successfully")
print(f" - ChromaDB: {persist_directory}")
print(f" - Collection: products ({collection.count()} documents)")
print(f" - Top K: {top_k}")
def get_embedding(text: str) -> List[float]:
"""Get embedding from Azure OpenAI"""
response = azure_client.embeddings.create(
input=text,
model=embedding_deployment
)
return response.data[0].embedding
def format_product_results(results: Dict) -> List[Dict]:
"""Format ChromaDB results into a clean list of products"""
products = []
for i, metadata in enumerate(results['metadatas'][0]):
product = {
"name": metadata['name'],
"category": metadata['category'],
"brand": metadata['brand'],
"price": metadata['price'],
"price_category": metadata['price_category'],
"description": metadata['description'],
"relevance_score": 1 - results['distances'][0][i] # Convert distance to similarity
}
products.append(product)
return products
@mcp.tool()
def search_products(query: str, limit: Optional[int] = None) -> List[Dict]:
"""
Search for similar products using natural language query.
Args:
query: Natural language search query (e.g., "red knife", "cheap mug", "waterproof tent")
limit: Maximum number of results to return (default: TOP_K from .env)
Returns:
List of products with their details and relevance scores
"""
if limit is None:
limit = top_k
# Get embedding for query
query_embedding = get_embedding(query)
# Search in ChromaDB
results = collection.query(
query_embeddings=[query_embedding],
n_results=min(limit, collection.count())
)
return format_product_results(results)
@mcp.tool()
def search_products_by_category(query: str, category: str, limit: Optional[int] = None) -> List[Dict]:
"""
Search for similar products within a specific category using natural language.
Args:
query: Natural language search query
category: Product category to filter by (e.g., "Tents", "Backpacks")
limit: Maximum number of results to return (default: TOP_K from .env)
Returns:
List of products matching the query within the specified category
"""
if limit is None:
limit = top_k
# Get embedding for query
query_embedding = get_embedding(query)
# Search with category filter
results = collection.query(
query_embeddings=[query_embedding],
n_results=min(limit, collection.count()),
where={"category": category}
)
return format_product_results(results)
@mcp.tool()
def search_products_by_brand(query: str, brand: str, limit: Optional[int] = None) -> List[Dict]:
"""
Search for similar products from a specific brand using natural language.
Args:
query: Natural language search query
brand: Product brand to filter by (e.g., "HikeMate", "OutdoorLiving")
limit: Maximum number of results to return (default: TOP_K from .env)
Returns:
List of products matching the query from the specified brand
"""
if limit is None:
limit = top_k
# Get embedding for query
query_embedding = get_embedding(query)
# Search with brand filter
results = collection.query(
query_embeddings=[query_embedding],
n_results=min(limit, collection.count()),
where={"brand": brand}
)
return format_product_results(results)
@mcp.tool()
def search_products_by_category_and_brand(
query: str,
category: str,
brand: str,
limit: Optional[int] = None
) -> List[Dict]:
"""
Search for similar products within a specific category and brand using natural language.
Args:
query: Natural language search query
category: Product category to filter by
brand: Product brand to filter by
limit: Maximum number of results to return (default: TOP_K from .env)
Returns:
List of products matching the query within the specified category and brand
"""
if limit is None:
limit = top_k
# Get embedding for query
query_embedding = get_embedding(query)
# Search with category and brand filters
results = collection.query(
query_embeddings=[query_embedding],
n_results=min(limit, collection.count()),
where={
"$and": [
{"category": category},
{"brand": brand}
]
}
)
return format_product_results(results)
@mcp.tool()
def get_categories() -> List[str]:
"""
Get all unique product categories available in the catalog.
Returns:
List of unique category names
"""
# Get all documents
all_docs = collection.get()
# Extract unique categories
categories = set()
for metadata in all_docs['metadatas']:
categories.add(metadata['category'])
return sorted(list(categories))
@mcp.tool()
def get_brands() -> List[str]:
"""
Get all unique product brands available in the catalog.
Returns:
List of unique brand names
"""
# Get all documents
all_docs = collection.get()
# Extract unique brands
brands = set()
for metadata in all_docs['metadatas']:
brands.add(metadata['brand'])
return sorted(list(brands))
@mcp.tool()
def get_product_description(product_name: str) -> Dict:
"""
Get complete product information by product name.
Args:
product_name: Exact or partial product name to search for
Returns:
Dictionary containing all product metadata (name, category, brand, price, description, etc.)
"""
# Search for product by name using contains filter
all_docs = collection.get()
# Find matching products
matching_products = []
for i, metadata in enumerate(all_docs['metadatas']):
if product_name.lower() in metadata['name'].lower():
matching_products.append(metadata)
if not matching_products:
return {"error": f"No product found matching '{product_name}'"}
if len(matching_products) == 1:
return matching_products[0]
else:
return {
"message": f"Found {len(matching_products)} products matching '{product_name}'",
"products": matching_products
}
@mcp.tool()
def search_products_with_price_filter(
query: str,
price_operator: str,
price_threshold: float,
limit: Optional[int] = None
) -> List[Dict]:
"""
Search for products using natural language and filter by price.
Note: ChromaDB has limited support for numeric comparisons, so we retrieve more results
and filter them post-query.
Args:
query: Natural language search query (e.g., "Hiking Clothing")
price_operator: Price comparison operator ("less_than", "greater_than", "less_or_equal", "greater_or_equal")
price_threshold: Price value to compare against (e.g., 50.0)
limit: Maximum number of results to return after filtering (default: TOP_K from .env)
Returns:
List of products matching the query and price criteria
"""
if limit is None:
limit = top_k
# Get embedding for query
query_embedding = get_embedding(query)
# Retrieve maximum results to ensure we have enough after filtering
max_results = collection.count()
# Search in ChromaDB (get all results for better filtering)
results = collection.query(
query_embeddings=[query_embedding],
n_results=max_results
)
# Filter by price
filtered_products = []
for i, metadata in enumerate(results['metadatas'][0]):
price = metadata['price']
# Apply price filter
include = False
if price_operator == "less_than" and price < price_threshold:
include = True
elif price_operator == "greater_than" and price > price_threshold:
include = True
elif price_operator == "less_or_equal" and price <= price_threshold:
include = True
elif price_operator == "greater_or_equal" and price >= price_threshold:
include = True
if include:
product = {
"name": metadata['name'],
"category": metadata['category'],
"brand": metadata['brand'],
"price": metadata['price'],
"price_category": metadata['price_category'],
"description": metadata['description'],
"relevance_score": 1 - results['distances'][0][i]
}
filtered_products.append(product)
# Stop if we have enough results
if len(filtered_products) >= limit:
break
return filtered_products[:limit]
# Initialize services on startup
initialize_services()
# Create ASGI application for HTTP deployment
app = mcp.http_app()
# For direct running (development)
if __name__ == "__main__":
print("\n" + "="*60)
print("Starting Product Recommendation MCP Server")
print("="*60)
mcp.run(transport="http", host="0.0.0.0", port=8000)