III. Solutions (Cookbook)

Personalized Recommendations

Overview

Build personalized content recommendation engines that analyze viewing history and generate tailored suggestions. This kind of application is good for streaming platforms, content apps, editorial tools, personalization engines, recommendation APIs, etc. This example shows how to:

• Analyze user viewing history to understand preferences
• Generate personalized recommendations with explanations
• Filter by available streaming services
• Provide conversational discovery experiences

Command-Line Recommendations

Generate recommendations in 15 minutes with a simple Python script.

Prerequisites

Before starting, complete the setup from Connecting to the MCP Server:

1. Dependencies installed: pip install litellm boto3 mcp python-dotenv
2. .env configured with your Cognito credentials and at least one LLM API key
3. mcp_utils.py copied into your project directory

Create Recommendation Engine

Create recommend_cli.py:

Code
 
import asyncio
import json
import os
import sys
from dotenv import load_dotenv
from litellm import completion

# Add shared folder to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'shared'))
from mcp_utils import GracenoteClient

load_dotenv()

DEFAULT_MODEL = "claude-sonnet-4-6"

SYSTEM_PROMPT = """
You are a personalized recommendation engine powered by Gracenote.

Your role:
- Analyze user viewing history to understand preferences
- Generate compelling, personalized recommendations
- Filter by available streaming services
- Explain why each recommendation fits the user

Rules:
1. Use resolve_entities to verify titles from viewing history
2. Analyze patterns: genres, cast, themes, release years
3. Use get_availability to filter by user's streaming services
4. Provide 5-10 recommendations ranked by relevance
5. Explain each recommendation in 1-2 sentences

Output format:
For each recommendation, include:
- Title and year
- Why it's recommended (based on viewing history)
- Where to watch (if filtering by services)
- Brief description
"""

async def get_recommendations(viewing_history, streaming_services=None, count=10, model=DEFAULT_MODEL):
    """Generate personalized recommendations."""

    async with GracenoteClient() as client:
        tools = await client.tools_for_litellm()

        prompt = f"""
Based on this viewing history:
{json.dumps(viewing_history, indent=2)}

Generate {count} personalized recommendations.
"""
        if streaming_services:
            prompt += f"\nFilter to content available on: {', '.join(streaming_services)}"

        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": prompt}
        ]

        response = completion(model=model, messages=messages, tools=tools, max_tokens=3000)
        response_message = response.choices[0].message

        while response_message.get("tool_calls"):
            messages.append(response_message.model_dump())

            for tool_call in response_message.tool_calls:
                tool_args = json.loads(tool_call.function.arguments)
                result = await client.call_tool(tool_call.function.name, tool_args)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "name": tool_call.function.name,
                    "content": str(result.content)
                })

            response = completion(model=model, messages=messages, tools=tools, max_tokens=3000)
            response_message = response.choices[0].message

        return response_message.content or "No recommendations generated"

async def main():
    # Example: User's viewing history
    viewing_history = [
        {"title": "Breaking Bad", "rating": 5},
        {"title": "The Wire", "rating": 5},
        {"title": "True Detective", "rating": 4},
        {"title": "Mindhunter", "rating": 4},
        {"title": "Ozark", "rating": 4}
    ]
    
    # Example: User's streaming services
    streaming_services = ["Netflix", "Hulu", "Max"]
    
    print("🎬 Generating personalized recommendations...\n")
    print(f"Based on viewing history: {', '.join([h['title'] for h in viewing_history])}")
    print(f"Available on: {', '.join(streaming_services)}\n")
    
    recommendations = await get_recommendations(
        viewing_history=viewing_history,
        streaming_services=streaming_services,
        count=10
    )
    
    print("=" * 80)
    print("🎯 PERSONALIZED RECOMMENDATIONS")
    print("=" * 80)
    print("\nBased on your viewing patterns, here are shows you might enjoy:\n")
    print(recommendations)

if __name__ == "__main__":
    asyncio.run(main())

Run It

Code
 
python3 recommend_cli.py

Output:

Code
 
🎬 Generating personalized recommendations...

Based on viewing history: Breaking Bad, The Wire, True Detective, Mindhunter, Ozark
Available on: Netflix, Hulu, Max

================================================================================
🎯 PERSONALIZED RECOMMENDATIONS
================================================================================

Based on your viewing patterns, here are shows you might enjoy:

1. **Narcos** (2015)
   Why: Like Breaking Bad and Ozark, this explores the drug trade with complex 
   characters and moral ambiguity. Features the same intense crime drama style 
   you enjoyed.
   Where: Netflix
   
2. **Better Call Saul** (2015)
   Why: Direct Breaking Bad spinoff with the same creators. If you rated Breaking 
   Bad 5 stars, you'll love this character-driven prequel.
   Where: Netflix

3. **The Sopranos** (1999)
   Why: Pioneered the complex antihero format that Breaking Bad and The Wire 
   perfected. Essential viewing for fans of character-driven crime dramas.
   Where: Max

[... 7 more recommendations ...]

Advanced: REST API Service

Build a production recommendation API with FastAPI.

File Structure

Code
 
personalized-recommendations/
├── recommend_api.py     # FastAPI application
├── mcp_utils.py         # Reusable MCP client (from Connecting to MCP)
├── requirements.txt     # Python dependencies
└── .env                 # Environment variables

recommend_api.py

Code
 
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import json
import os
import sys
from typing import List, Optional
from dotenv import load_dotenv
from litellm import completion

# Add shared folder to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'shared'))
from mcp_utils import GracenoteClient

load_dotenv()

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

DEFAULT_MODEL = "claude-sonnet-4-6"

SYSTEM_PROMPT = """
You are a personalized recommendation engine powered by Gracenote.
Analyze user viewing history and generate compelling recommendations.
Use MCP tools to verify titles and check availability.
Provide clear explanations for each recommendation.
"""

class ViewingItem(BaseModel):
    title: str
    rating: Optional[int] = None

class RecommendationRequest(BaseModel):
    viewing_history: List[ViewingItem]
    streaming_services: Optional[List[str]] = None
    count: int = 10
    model: Optional[str] = DEFAULT_MODEL

class RecommendationResponse(BaseModel):
    recommendations: str

@app.post("/recommend", response_model=RecommendationResponse)
async def recommend(request: RecommendationRequest):
    try:
        async with GracenoteClient() as client:
            tools = await client.tools_for_litellm()

            prompt = f"Based on viewing history: {[h.title for h in request.viewing_history]}\n"
            prompt += f"Generate {request.count} personalized recommendations.\n"
            if request.streaming_services:
                prompt += f"Filter to: {', '.join(request.streaming_services)}"

            messages = [
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": prompt}
            ]

            response = completion(model=request.model, messages=messages, tools=tools, max_tokens=2048)
            response_message = response.choices[0].message

            while response_message.get("tool_calls"):
                messages.append(response_message.model_dump())

                for tool_call in response_message.tool_calls:
                    tool_args = json.loads(tool_call.function.arguments)
                    result = await client.call_tool(tool_call.function.name, tool_args)
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "name": tool_call.function.name,
                        "content": str(result.content)
                    })

                response = completion(model=request.model, messages=messages, tools=tools, max_tokens=2048)
                response_message = response.choices[0].message

            return RecommendationResponse(recommendations=response_message.content or "")

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health():
    return {"status": "healthy"}

requirements.txt

Code
 
litellm>=1.40.0
fastapi>=0.104.0
uvicorn>=0.24.0
boto3>=1.35.0
mcp>=1.0.0
python-dotenv>=1.0.0

Running the API

Code
 
pip install -r requirements.txt
uvicorn recommend_api:app --host 0.0.0.0 --port 8001

Testing the API

Code
 
curl -X POST http://localhost:8001/recommend \
  -H "Content-Type: application/json" \
  -d '{
    "viewing_history": [
      {"title": "Breaking Bad", "rating": 5},
      {"title": "The Wire", "rating": 5}
    ],
    "streaming_services": ["Netflix", "Hulu"],
    "count": 5
  }'

Production Best Practices

Caching Recommendations

Cache recommendations to reduce API calls:

Code
 
import redis
import hashlib
import json

cache = redis.Redis(host='localhost', port=6379, decode_responses=True)

def cache_key(viewing_history, services):
    data = json.dumps({"history": viewing_history, "services": services}, sort_keys=True)
    return hashlib.md5(data.encode()).hexdigest()

async def get_cached_recommendations(viewing_history, services):
    key = cache_key(viewing_history, services)
    cached = cache.get(key)
    if cached:
        return json.loads(cached)
    
    # Generate recommendations
    recommendations = await get_recommendations(viewing_history, services)
    
    # Cache for 1 hour
    cache.setex(key, 3600, json.dumps(recommendations))
    return recommendations

A/B Testing

Test different recommendation strategies:

Code
 
import random

def get_recommendation_strategy(user_id):
    # Assign users to different strategies
    strategies = ["collaborative", "content_based", "hybrid"]
    return strategies[hash(user_id) % len(strategies)]

async def recommend_with_strategy(user_id, viewing_history):
    strategy = get_recommendation_strategy(user_id)
    
    if strategy == "collaborative":
        prompt = "Use collaborative filtering patterns..."
    elif strategy == "content_based":
        prompt = "Focus on content similarity..."
    else:
        prompt = "Combine both approaches..."
    
    return await get_recommendations(viewing_history, prompt=prompt)

Monitoring and Analytics

Track recommendation performance:

Code
 
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

async def recommend_with_tracking(user_id, viewing_history):
    start = datetime.now()
    
    try:
        recommendations = await get_recommendations(viewing_history)
        
        duration = (datetime.now() - start).total_seconds()
        logger.info(f"Recommendations generated for user {user_id} in {duration:.2f}s")
        
        # Track to analytics
        track_event("recommendation_generated", {
            "user_id": user_id,
            "duration": duration,
            "history_size": len(viewing_history)
        })
        
        return recommendations
    except Exception as e:
        logger.error(f"Recommendation failed for user {user_id}: {e}")
        raise

Deployment

Docker

Code
 
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .

CMD ["uvicorn", "recommend_api:app", "--host", "0.0.0.0", "--port", "8001"]

Kubernetes

Code
 
apiVersion: apps/v1
kind: Deployment
metadata:
  name: recommendation-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: recommendation-api
  template:
    metadata:
      labels:
        app: recommendation-api
    spec:
      containers:
      - name: api
        image: your-registry/recommendation-api:latest
        ports:
        - containerPort: 8001
        env:
        - name: MCP_SERVER_HOST
          valueFrom:
            secretKeyRef:
              name: gracenote-creds
              key: host

See also: Catalog Enrichment | Search & Discovery | Best Practices | Tool Reference