AI and MCP (Model Context Protocol) integration patterns for RawDrive. Use when working with the ai-service, implementing AI features, or building MCP tools.
RawDrive's AI Service provides:
The service uses configurable LLM providers via environment variables. Never hardcode provider names or API keys.
| Purpose | Location |
|---|---|
| FastAPI entry | services/ai-service/src/main.py |
| MCP server | services/ai-service/src/mcp/server.py |
| Face detection | backend/src/app/services/face_detection_service.py |
| AI providers | backend/src/app/services/ai_providers/ |
| Content detection | backend/src/app/services/content_detection_service.py |
# services/ai-service/src/mcp/server.py
from typing import Annotated
from fastmcp import FastMCP
mcp_server = FastMCP(name="rawdrive-ai", version="1.0.0")
@mcp_server.tool()
async def detect_faces(
photo_id: Annotated[str, "Photo ID in database"],
workspace_id: Annotated[str, "Workspace ID for isolation"],
detect_attributes: Annotated[bool, "Detect age/gender"] = True,
) -> dict:
"""
Detect faces in a photo and generate embeddings.
Returns bounding boxes, 512-dim embeddings, and attributes.
"""
service = FaceRecognitionService()
result = await service.detect_faces(photo_id, workspace_id)
return {
"photo_id": result.photo_id,
"face_count": len(result.faces),
"faces": [f.to_dict() for f in result.faces],
}
# backend/src/app/services/ai_providers/base.py
class AIProvider:
"""Base class for AI providers - configured via environment."""
def __init__(self):
# Load from env - NEVER hardcode
self.provider = os.environ.get("AI_PROVIDER")
self.api_key = os.environ.get("AI_API_KEY")
self.model = os.environ.get("AI_MODEL")
async def analyze_photo(self, image: bytes, asset_id: str) -> dict:
"""Analyze photo using configured provider."""
raise NotImplementedError
# backend/src/app/services/face_detection_service.py
class FaceDetectionService:
"""Multi-provider face detection with failover."""
async def detect_faces(
self,
workspace_id: UUID,
asset_id: UUID,
) -> list[DetectedFace]:
"""
Detect faces using provider chain:
1. Cloud Vision (primary)
2. Gemini (fallback)
3. Local DeepFace (last resort)
"""
for provider in self._providers:
try:
return await provider.detect(workspace_id, asset_id)
except ProviderError:
continue
raise AllProvidersFailedError()
# Uses CLIP embeddings stored in pgvector
async def search(query: str, workspace_id: UUID) -> list[SearchResult]:
# Encode query
embedding = clip_model.encode(query)
# Search pgvector
results = await db.execute("""
SELECT asset_id, 1 - (embedding <=> $1::vector) as score
FROM asset_embeddings
WHERE workspace_id = $2
ORDER BY embedding <=> $1::vector
LIMIT 20
""", [embedding, workspace_id])
return results
// frontend/src/services/aiService.ts
class AIService {
async analyzePhoto(assetId: string): Promise<PhotoAnalysis> {
// Proxy through backend - never expose API keys to frontend
return api.post('/api/v1/ai/analyze', { assetId });
}
async searchPhotos(query: string, galleryId?: string): Promise<SearchResult[]> {
return api.post('/api/v1/ai/search', { query, galleryId });
}
async detectFaces(assetId: string): Promise<FaceResult[]> {
return api.post('/api/v1/ai/faces/detect', { assetId });
}
}
# Cache by image hash to avoid reprocessing
async def analyze_with_cache(image_hash: str, image: bytes):
cache_key = f"ai:analysis:{image_hash}"
if cached := await redis.get(cache_key):
return json.loads(cached)
result = await ai_provider.analyze(image)
await redis.setex(cache_key, 86400, json.dumps(result))
return result
# Use local models for embeddings (cheaper)
# Use LLM for complex understanding (captions, quality)