In the rapidly evolving landscape of AI-powered applications, Model Context Protocol (MCP) servers have emerged as a critical infrastructure component for connecting large language models to external data sources and tools. As organizations scale their AI deployments from prototypes to production systems handling thousands of concurrent requests, understanding the architectural decisions and operational considerations becomes paramount. This comprehensive guide walks through the complete journey of deploying and scaling an MCP server, drawing from real-world implementation experience across multiple industry verticals.
The Challenge: Scaling AI Infrastructure Under Pressure
Three months into our e-commerce platform's AI customer service integration, we faced a critical inflection point. Our initial deployment—a single MCP server instance handling product inquiries, order tracking, and return requests—was crumbling under Black Friday traffic. Response times had spiked from 800ms to over 4 seconds, error rates exceeded 15%, and our customer satisfaction scores were plummeting. The engineering team was spending more time firefighting than building new features.
I remember the war room meeting where we decided to fundamentally rethink our MCP server architecture. We had two options: throw money at bigger instances or build a truly scalable system. This guide documents the architectural decisions, implementation patterns, and hard-won lessons from that transformation journey—covering everything from single-server deployments for indie developers to enterprise-grade systems serving millions of requests daily.
Understanding MCP Server Architecture Fundamentals
Before diving into scaling strategies, we need to establish a clear mental model of how MCP servers operate within the broader AI application ecosystem. An MCP server acts as an intermediary layer that standardizes how AI models interact with external tools, data sources, and services. Unlike traditional API integrations where each application handles its own connection logic, MCP provides a unified protocol that abstracts away the complexity of tool invocation, context management, and response formatting.
The MCP Protocol Stack
At its core, the MCP protocol operates through a request-response cycle augmented by context management capabilities. When an AI model determines it needs to use a tool—for example, fetching real-time inventory levels for a product inquiry—it sends a tool invocation request to the MCP server. The server authenticates the request, executes the underlying operation, and returns formatted results that the model can incorporate into its response.
# Core MCP Server Architecture Pattern
Demonstrates the fundamental request-response cycle
import asyncio
import json
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
import hashlib
@dataclass
class MCPContext:
"""Manages conversation context and tool state across requests."""
session_id: str
created_at: datetime = field(default_factory=datetime.now)
tool_invocations: List[Dict[str, Any]] = field(default_factory=list)
cache: Dict[str, Any] = field(default_factory=dict)
metadata: Dict[str, Any] = field(default_factory=dict)
def add_invocation(self, tool_name: str, params: Dict, result: Any):
self.tool_invocations.append({
"tool": tool_name,
"params": params,
"result": result,
"timestamp": datetime.now().isoformat()
})
def get_cache_key(self, tool: str, params: Dict) -> str:
"""Generate consistent cache key for request deduplication."""
param_str = json.dumps(params, sort_keys=True)
return hashlib.sha256(f"{tool}:{param_str}".encode()).hexdigest()[:16]
class MCPServerCore:
"""
Core MCP server implementation with context management
and basic caching capabilities.
"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.contexts: Dict[str, MCPContext] = {}
self.tool_registry: Dict[str, callable] = {}
self.cache_ttl = config.get("cache_ttl_seconds", 300)
async def handle_request(
self,
session_id: str,
tool_name: str,
parameters: Dict[str, Any],
use_cache: bool = True
) -> Dict[str, Any]:
"""
Process incoming MCP tool invocation request.
Handles context management, caching, and tool routing.
"""
# Initialize or retrieve session context
if session_id not in self.contexts:
self.contexts[session_id] = MCPContext(session_id=session_id)
context = self.contexts[session_id]
# Check cache if enabled
if use_cache:
cache_key = context.get_cache_key(tool_name, parameters)
if cache_key in context.cache:
cached_result = context.cache[cache_key]
cached_age = (datetime.now() - cached_result["cached_at"]).total_seconds()
if cached_age < self.cache_ttl:
return {
"success": True,
"cached": True,
"age_seconds": cached_age,
"result": cached_result["data"]
}
# Validate and execute tool
if tool_name not in self.tool_registry:
return {
"success": False,
"error": f"Unknown tool: {tool_name}",
"available_tools": list(self.tool_registry.keys())
}
try:
start_time = datetime.now()
result = await self.tool_registry[tool_name](**parameters)
execution_time = (datetime.now() - start_time).total_seconds()
# Update context and cache
context.add_invocation(tool_name, parameters, result)
if use_cache:
context.cache[cache_key] = {
"data": result,
"cached_at": datetime.now()
}
return {
"success": True,
"result": result,
"execution_time_ms": round(execution_time * 1000, 2),
"tool_invocations_count": len(context.tool_invocations)
}
except Exception as e:
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__
}
def register_tool(self, name: str, handler: callable):
"""Register a new tool with the MCP server."""
self.tool_registry[name] = handler
print(f"Registered tool: {name}")
Example tool implementations
async def get_product_inventory(product_id: str, location: Optional[str] = None) -> Dict:
"""Fetch current inventory levels for a product."""
# Simulated inventory lookup
return {
"product_id": product_id,
"available": True,
"quantity": 142,
"locations": ["warehouse_east", "warehouse_west"],
"estimated_restock": "2024-11-20T14:00:00Z"
}
async def check_order_status(order_id: str, email: str) -> Dict:
"""Retrieve current status of a customer order."""
# Simulated order lookup
return {
"order_id": order_id,
"status": "shipped",
"tracking_number": "1Z999AA10123456784",
"estimated_delivery": "2024-11-18",
"last_update": "2024-11-15T09:23:00Z"
}
Initialize and configure server
config = {
"cache_ttl_seconds": 300,
"max_contexts": 10000,
"request_timeout_seconds": 30
}
server = MCPServerCore(config)
server.register_tool("get_inventory", get_product_inventory)
server.register_tool("check_order_status", check_order_status)
print("MCP Server initialized with core architecture")Deployment Patterns: From Single Instance to Global Scale
The journey to scalable MCP server infrastructure follows a well-documented progression. Most teams start with a single-instance deployment—appropriate for development environments and early-stage products with limited traffic. As usage grows, they transition through vertical scaling, horizontal scaling with load balancers, and eventually reach distributed architectures capable of handling millions of requests across multiple geographic regions.
Pattern 1: Single Instance Deployment for Indie Developers
For indie developers and small teams just beginning their MCP journey, a single-instance deployment provides the fastest path from zero to working prototype. This approach minimizes operational complexity while providing sufficient performance for applications with up to 1,000 daily active users. The key advantage is simplicity: you deploy one server, configure one set of credentials, and manage one deployment pipeline.
# Complete Single-Instance MCP Server Deployment
Production-ready setup for indie developers and small teams
Uses HolySheep AI for LLM inference with <50ms latency
import os
import asyncio
import logging
from typing import Dict, Any, Optional
from dataclasses import dataclass
import httpx
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
Configuration
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Logging setup
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp_server")
@dataclass
class ServerConfig:
"""Centralized configuration for MCP server."""
host: str = "0.0.0.0"
port: int = 8000
workers: int = 4
max_request_size_mb: int = 10
timeout_seconds: int = 120
rate_limit_per_minute: int = 60
cache_enabled: bool = True
cache_ttl_seconds: int = 300
class ChatRequest(BaseModel):
"""Incoming chat request model."""
session_id: str
message: str
context: Optional[Dict[str, Any]] = None
model: str = "gpt-4.1"
temperature: float = 0.7
max_tokens: int = 1000
class ToolInvocation(BaseModel):
"""MCP tool invocation request."""
tool_name: str
parameters: Dict[str, Any]
class HolySheepAIClient:
"""
Production client for HolySheep AI API.
Handles authentication, retry logic, and response parsing.
"""
def __init__(self, api_key: str, base_url: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.base_url = base_url
self.timeout = httpx.Timeout(120.0, connect=10.0)
self.retry_attempts = 3
self.retry_delay = 1.0
async def chat_completion(
self,
messages: list,
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
Send chat completion request to HolySheep AI.
Supports GPT-4.1 at $8/MTok with <50ms infrastructure latency.
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
async with httpx.AsyncClient(timeout=self.timeout) as client:
for attempt in range(self.retry_attempts):
try:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
# Rate limit exceeded - implement backoff
wait_time = self.retry_delay * (2 ** attempt)
logger.warning(f"Rate limited. Waiting {wait_time}s before retry.")
await asyncio.sleep(wait_time)
else:
raise HTTPException(
status_code=e.response.status_code,
detail=f"API error: {str(e)}"
)
except httpx.RequestError as e:
if attempt < self.retry_attempts - 1:
await asyncio.sleep(self.retry_delay)
continue
raise HTTPException(
status_code=503,
detail=f"Service unavailable: {str(e)}"
)
raise HTTPException(status_code=500, detail="Max retries exceeded")
Tool implementations for e-commerce customer service
TOOLS = {
"get_product_info": {
"description": "Fetch product details including price, availability, and specifications",
"handler": lambda pid: {
"product_id": pid,
"name": "Premium Wireless Headphones",
"price": 149.99,
"currency": "USD",
"in_stock": True,
"rating": 4.7
}
},
"check_inventory": {
"description": "Check real-time inventory levels across warehouse locations",
"handler": lambda pid, loc=None: {
"product_id": pid,
"total_available": 342,
"locations": ["east", "west", "central"],
"next_restock": "2024-12-01"
}
},
"track_order": {
"description": "Get order status and shipping information",
"handler": lambda oid: {
"order_id": oid,
"status": "shipped",
"tracking": "1Z999AA10123456784",
"eta": "2024-11-20"
}
},
"process_return": {
"description": "Initiate return request and generate label",
"handler": lambda oid, reason: {
"return_id": f"RET-{oid[-8:]}",
"label_url": "https://returns.example.com/label.pdf",
"instructions": "Ship within 5 business days"
}
}
}
FastAPI application setup
app = FastAPI(title="E-Commerce MCP Server", version="1.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"]
)
ai_client = HolySheepAIClient(api_key=HOLYSHEEP_API_KEY)
config = ServerConfig()
@app.post("/mcp/chat")
async def handle_chat(request: ChatRequest, background_tasks: BackgroundTasks):
"""
Main MCP chat endpoint.
Integrates with HolySheep AI for LLM inference.
"""
# Construct messages with optional context
system_prompt = """You are an expert customer service assistant for an e-commerce platform.
Use the available tools to help customers with:
- Product information and recommendations
- Order status and tracking
- Returns and exchanges
- General inquiries
Always be helpful, accurate, and concise."""
messages = [
{"role": "system", "content": system_prompt}
]
if request.context:
context_str = "\n".join([f"{k}: {v}" for k, v in request.context.items()])
messages.append({
"role": "system",
"content": f"Session context:\n{context_str}"
})
messages.append({"role": "user", "content": request.message})
# Call HolySheep AI
try:
response = await ai_client.chat_completion(
messages=messages,
model=request.model,
temperature=request.temperature,
max_tokens=request.max_tokens
)
return {
"success": True,
"session_id": request.session_id,
"response": response["choices"][0]["message"]["content"],
"model": response.get("model", request.model),
"usage": response.get("usage", {})
}
except HTTPException:
raise
except Exception as e:
logger.error(f"Chat error: {str(e)}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/mcp/tools/invoke")
async def invoke_tool(invocation: ToolInvocation):
"""Direct tool invocation endpoint for MCP clients."""
if invocation.tool_name not in TOOLS:
raise HTTPException(
status_code=404,
detail=f"Tool '{invocation.tool_name}' not found"
)
tool = TOOLS[invocation.tool_name]
result = tool["handler"](**invocation.parameters)
return {
"success": True,
"tool": invocation.tool_name,
"result": result
}
@app.get("/mcp/tools")
async def list_tools():
"""List all available MCP tools."""
return {
"tools": [
{"name": name, "description": info["description"]}
for name, info in TOOLS.items()
]
}
@app.get("/health")
async def health_check():
"""Health check endpoint for load balancers."""
return {
"status": "healthy",
"version": "1.0.0",
"tools_registered": len(TOOLS)
}
if __name__ == "__main__":
logger.info(f"Starting MCP Server on {config.host}:{config.port}")
uvicorn.run(
app,
host=config.host,
port=config.port,
workers=config.workers
)When we deployed our initial single-instance setup for the e-commerce platform, we underestimated the traffic patterns that would emerge. Within the first week, we learned several critical lessons about capacity planning, even for modest deployments. Our monitoring revealed that most traffic clustered around predictable hours, with sharp spikes during flash sales and promotional events. This observation became the foundation for our autoscaling strategy.
Pattern 2: Horizontal Scaling with Load Balancing
As traffic grows beyond what a single instance can handle, horizontal scaling becomes necessary. This approach involves deploying multiple MCP server instances behind a load balancer, distributing incoming requests across the fleet. The complexity increases—now you must manage state synchronization, session affinity, and health monitoring—but the scalability gains are substantial.
# Horizontal Scaling Architecture with Health Monitoring
Production-ready multi-instance deployment for growing applications
import os
import asyncio
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
import hashlib
import json
import redis.asyncio as redis
from fastapi import FastAPI, HTTPException, Header, Query
from fastapi.middleware.cors import CORSMiddleware
import httpx
import uvicorn
from pydantic import BaseModel
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("scaled_mcp")
Configuration
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
@dataclass
class InstanceMetrics:
"""Metrics for a single MCP server instance."""
instance_id: str
host: str
port: int
healthy: bool = True
current_requests: int = 0
total_requests: int = 0
avg_response_time_ms: float = 0.0
error_rate: float = 0.0
last_health_check: datetime = field(default_factory=datetime.now)
registered_at: datetime = field(default_factory=datetime.now)
class ScalingOrchestrator:
"""
Manages MCP server fleet with automatic scaling and health monitoring.
Implements least-connections load balancing with health-aware routing.
"""
def __init__(
self,
redis_client: redis.Redis,
max_instances: int = 10,
min_instances: int = 2,
scale_up_threshold: float = 0.7,
scale_down_threshold: float = 0.3
):
self.redis = redis_client
self.instances: Dict[str, InstanceMetrics] = {}
self.max_instances = max_instances
self.min_instances = min_instances
self.scale_up_threshold = scale_up_threshold
self.scale_down_threshold = scale_down_threshold
self.health_check_interval = 10 # seconds
self.response_times: Dict[str, List[float]] = defaultdict(list)
self.error_counts: Dict[str, int] = defaultdict(int)
self.request_counts: Dict[str, int] = defaultdict(int)
async def register_instance(
self,
instance_id: str,
host: str,
port: int
) -> bool:
"""Register a new MCP server instance with the orchestrator."""
if instance_id in self.instances:
logger.warning(f"Instance {instance_id} already registered")
return False
instance = InstanceMetrics(
instance_id=instance_id,
host=host,
port=port
)
self.instances[instance_id] = instance
# Store in Redis for cross-process coordination
await self.redis.hset(
"mcp_instances",
instance_id,
json.dumps({
"host": host,
"port": port,
"registered_at": datetime.now().isoformat()
})
)
logger.info(f"Registered instance {instance_id} at {host}:{port}")
return True
async def deregister_instance(self, instance_id: str) -> bool:
"""Remove an instance from the active fleet."""
if instance_id not in self.instances:
return False
del self.instances[instance_id]
await self.redis.hdel("mcp_instances", instance_id)
# Clean up metrics
self.response_times.pop(instance_id, None)
self.error_counts.pop(instance_id, None)
self.request_counts.pop(instance_id, None)
logger.info(f"Deregistered instance {instance_id}")
return True
async def health_check_instance(self, instance_id: str) -> bool:
"""Perform health check on a single instance."""
if instance_id not in self.instances:
return False
instance = self.instances[instance_id]
health_url = f"http://{instance.host}:{instance.port}/health"
try:
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(health_url)
if response.status_code == 200:
instance.healthy = True
instance.last_health_check = datetime.now()
return True
else:
logger.warning(
f"Instance {instance_id} returned status {response.status_code}"
)
except Exception as e:
logger.error(f"Health check failed for {instance_id}: {str(e)}")
instance.healthy = False
return False
async def periodic_health_checks(self):
"""Background task to continuously monitor instance health."""
while True:
for instance_id in list(self.instances.keys()):
await self.health_check_instance(instance_id)
await asyncio.sleep(self.health_check_interval)
async def record_request_metrics(
self,
instance_id: str,
response_time_ms: float,
is_error: bool = False
):
"""Record performance metrics for an instance."""
# Store recent response times (keep last 100)
self.response_times[instance_id].append(response_time_ms)
if len(self.response_times[instance_id]) > 100:
self.response_times[instance_id].pop(0)
# Update error count
if is_error:
self.error_counts[instance_id] += 1
# Update instance metrics
if instance_id in self.instances:
instance = self.instances[instance_id]
instance.avg_response_time_ms = sum(
self.response_times[instance_id]
) / len(self.response_times[instance_id])
instance.error_rate = (
self.error_counts[instance_id] /
max(1, sum(self.request_counts.values()))
)
instance.total_requests += 1
def select_instance(self) -> Optional[Tuple[str, InstanceMetrics]]:
"""
Select the best instance using least-connections algorithm.
Prefers healthy instances with lower load.
"""
healthy_instances = [
(id, inst) for id, inst in self.instances.items()
if inst.healthy
]
if not healthy_instances:
return None
# Sort by current request load
healthy_instances.sort(key=lambda x: x[1].current_requests)
return healthy_instances[0]
async def route_request(
self,
session_id: str,
request_payload: Dict
) -> Dict:
"""
Route request to optimal MCP server instance.
Implements consistent hashing for session affinity.
"""
# Generate consistent key for session-based routing
session_hash = int(
hashlib.md5(session_id.encode()).hexdigest()[:8],
16
)
# Get healthy instances
healthy_instances = [
inst for inst in self.instances.values() if inst.healthy
]
if not healthy_instances:
raise HTTPException(
status_code=503,
detail="No healthy instances available"
)
# Consistent hash to select instance
instance_idx = session_hash % len(healthy_instances)
selected = healthy_instances[instance_idx]
# Send request to selected instance
instance_url = f"http://{selected.host}:{selected.port}/mcp/chat"
start_time = datetime.now()
try:
async with httpx.AsyncClient(timeout=120.0) as client:
response = await client.post(
instance_url,
json=request_payload
)
response.raise_for_status()
# Record metrics
response_time = (
datetime.now() - start_time
).total_seconds() * 1000
await self.record_request_metrics(
selected.instance_id,
response_time,
is_error=False
)
return response.json()
except Exception as e:
await self.record_request_metrics(
selected.instance_id,
0,
is_error=True
)
raise HTTPException(
status_code=502,
detail=f"Instance error: {str(e)}"
)
async def get_fleet_status(self) -> Dict:
"""Get current status of the entire MCP server fleet."""
total_requests = sum(inst.total_requests for inst in self.instances.values())
healthy_count = sum(1 for inst in self.instances.values() if inst.healthy)
return {
"total_instances": len(self.instances),
"healthy_instances": healthy_count,
"total_requests": total_requests,
"instances": [
{
"id": inst.instance_id,
"healthy": inst.healthy,
"current_load": inst.current_requests,
"avg_response_time_ms": round(inst.avg_response_time_ms, 2),
"error_rate": round(inst.error_rate * 100, 2)
}
for inst in self.instances.values()
]
}
FastAPI Application with Scaling Orchestrator
app = FastAPI(title="Scaled MCP Server Fleet", version="2.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"]
)
redis_client: Optional[redis.Redis] = None
orchestrator: Optional[ScalingOrchestrator] = None
@app.on_event("startup")
async def startup():
"""Initialize Redis connection and orchestrator on startup."""
global redis_client, orchestrator
redis_client = await redis.from_url(REDIS_URL)
orchestrator = ScalingOrchestrator(
redis_client=redis_client,
max_instances=10,
min_instances=2
)
# Start background health check task
asyncio.create_task(orchestrator.periodic_health_checks())
logger.info("Scaled MCP orchestrator initialized")
@app.on_event("shutdown")
async def shutdown():
"""Cleanup on shutdown."""
if redis_client:
await redis_client.close()
class ChatRequest(BaseModel):
session_id: str
message: str
model: str = "gpt-4.1"
@app.post("/mcp/chat")
async def scaled_chat(request: ChatRequest):
"""Route chat request to optimal instance."""
payload = {
"session_id": request.session_id,
"message": request.message,
"model": request.model
}
return await orchestrator.route_request(
request.session_id,
payload
)
@app.post("/fleet/register")
async def register_instance(
instance_id: str = Query(...),
host: str = Query(...),
port: int = Query(...)
):
"""Register a new MCP server instance."""
success = await orchestrator.register_instance(instance_id, host, port)
return {"success": success, "instance_id": instance_id}
@app.post("/fleet/deregister")
async def deregister_instance(instance_id: str = Query(...)):
"""Deregister an MCP server instance."""
success = await orchestrator.deregister_instance(instance_id)
return {"success": success}
@app.get("/fleet/status")
async def fleet_status():
"""Get current fleet status."""
return await orchestrator.get_fleet_status()
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)Enterprise RAG System Implementation
For enterprise deployments handling sensitive documents and knowledge bases, the scaling considerations multiply significantly. A Retrieval-Augmented Generation (RAG) system built on MCP architecture must balance query latency, retrieval accuracy, and data freshness while maintaining strict security boundaries. In our enterprise implementation, we processed over 10 million documents across 50 knowledge bases, with query volumes exceeding 100,000 requests per day.
The architectural pattern that emerged from this implementation separates concerns across three distinct layers: the ingestion pipeline for document processing, the retrieval layer for semantic search, and the inference layer for response generation. Each layer scales independently based on its specific workload characteristics, and the MCP server acts as the orchestration layer coordinating across them.
# Enterprise RAG MCP Server with Vector Search Integration
Handles document ingestion, semantic retrieval, and context synthesis
Optimized for production workloads with $0.42/MTok on DeepSeek V3.2
import os
import asyncio
import hashlib
import logging
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import numpy as np
import httpx
from fastapi import FastAPI, HTTPException, BackgroundTasks, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("enterprise_rag")
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Embedding and inference pricing comparison
MODEL_PRICING = {
"gpt-4.1": {"input": 2.50, "output": 8.00, "currency": "USD"},
"claude-sonnet-4.5": {"input": 3.00, "output": 15.00, "currency": "USD"},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50, "currency": "USD"},
"deepseek-v3.2": {"input": 0.07, "output": 0.42, "currency": "USD"}
}
class DocumentStatus(Enum):
"""Document processing status tracking."""
PENDING = "pending"
PROCESSING = "processing"
INDEXED = "indexed"
FAILED = "failed"
ARCHIVED = "archived"
@dataclass
class Document:
"""Represents a document in the RAG system."""
doc_id: str
content: str
metadata: Dict[str, Any] = field(default_factory=dict)
status: DocumentStatus = DocumentStatus.PENDING
chunk_ids: List[str] = field(default_factory=list)
embedding_ids: List[str] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.now)
indexed_at: Optional[datetime] = None
@dataclass
class RetrievalResult:
"""Result from semantic search retrieval."""
doc_id: str
chunk_id: str
content: str
score: float
metadata: Dict[str, Any]
class VectorStore:
"""
Simplified vector store for document embeddings.
In production, replace with Pinecone, Weaviate, or pgvector.
"""
def __init__(self, dimension: int = 1536):
self.dimension = dimension
self.vectors: Dict[str, np.ndarray] = {}
self.metadata: Dict[str, Dict] = {}
def add_vector(
self,
vector_id: str,
embedding: np.ndarray,
metadata: Dict = None
):
"""Add embedding vector to store."""
self.vectors[vector_id] = embedding
self.metadata[vector_id] = metadata or {}
def search(
self,
query_embedding: np.ndarray,
top_k: int = 5,
min_score: float = 0.7
) -> List[Tuple[str, float]]:
"""Perform cosine similarity search."""
scores = []
for vec_id, vec in self.vectors.items():
# Cosine similarity
similarity = np.dot(query_embedding, vec) / (
np.linalg.norm(query_embedding) * np.linalg.norm(vec)
)
if similarity >= min_score:
scores.append((vec_id, float(similarity)))
# Sort by score descending
scores.sort(key=lambda x: x[1], reverse=True)
return scores[:top_k]
class EmbeddingService:
"""
Manages document embedding generation.
Uses HolySheep AI for efficient embedding computation.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.batch_size = 32
self.dimension = 1536
async def embed_texts(self, texts: List[str]) -> List[np.ndarray]:
"""
Generate embeddings for multiple texts.
Batches requests for efficiency.
"""
embeddings = []
for i in range(0, len(texts), self.batch_size):
batch = texts[i:i + self.batch_size]
# Simulated embedding (in production, call embedding API)
batch_embeddings = []
for text in batch:
# Simple hash-based simulation for demo
hash_val = int(hashlib.md5(text.encode()).hexdigest()[:8], 16)
np.random.seed(hash_val % (2**31))
embedding = np.random.randn(self.dimension)
embedding = embedding / np.linalg.norm(embedding)
batch_embeddings.append(embedding)
embeddings.extend(batch_embeddings)
return embeddings
async def embed_query(self, query: str) -> np.ndarray:
"""Generate embedding for search query."""
embeddings = await self.embed_texts([query])
return embeddings[0]
class