Custom AI Backends with MCP

Building specialized AI services and backends that can be accessed as standard MCP servers, enabling clients to interact with custom ML models, domain-specific AI, and proprietary algorithms.

Overview

MCP allows you to wrap any AI or ML service as a standard protocol interface. Custom backends expose specialized capabilities through tools and resources, making them accessible to any MCP client.

Custom LLM Service

from mcp.server import Server, Tool
from mcp.types import TextContent
from transformers import pipeline
import json

server = Server("domain-llm-backend")

# Load specialized models
legal_qa = pipeline(
    "question-answering",
    model="deepset/roberta-base-squad2"
)

legal_summarizer = pipeline(
    "summarization",
    model="facebook/bart-large-cnn"
)

# Legal domain context
legal_documents = {
    "contract-101": "This contract outlines the terms of service between parties...",
    "policy-201": "The refund policy states that customers have 30 days to return items..."
}

@server.tool(
    name="query_legal_documents",
    description="Query legal documents with question-answering"
)
def query_legal_documents(question: str, document_id: str = None) -> str:
    """Answer questions about legal documents"""
    try:
        if document_id and document_id in legal_documents:
            context = legal_documents[document_id]
        else:
            context = " ".join(legal_documents.values())
        
        result = legal_qa(question=question, context=context)
        
        return TextContent(text=json.dumps({
            "answer": result["answer"],
            "confidence": float(result["score"]),
            "context": result.get("context", "")[:500]
        }))
    
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

@server.tool(
    name="summarize_legal_text",
    description="Summarize legal documents"
)
def summarize_legal_text(document_id: str, length: str = "medium") -> str:
    """Summarize legal document"""
    try:
        if document_id not in legal_documents:
            return TextContent(text="Document not found")
        
        text = legal_documents[document_id]
        
        # Adjust max_length based on requested length
        max_length = {"short": 50, "medium": 100, "long": 200}[length]
        min_length = max(10, max_length // 3)
        
        result = legal_summarizer(
            text,
            max_length=max_length,
            min_length=min_length,
            do_sample=False
        )
        
        return TextContent(text=result[0]["summary_text"])
    
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

Computer Vision Backend

from mcp.server import Server
from mcp.types import TextContent
import base64
import json
from PIL import Image
from io import BytesIO
import numpy as np

server = Server("vision-backend")

class ImageAnalyzer:
    def __init__(self):
        # In practice, would load actual models
        self.object_detector = None
        self.ocr_engine = None
    
    def detect_objects(self, image_data: str):
        """Detect objects in image"""
        try:
            image = Image.open(BytesIO(base64.b64decode(image_data)))
            
            # Simulated detection
            detections = [
                {"class": "person", "confidence": 0.95, "bbox": [10, 20, 100, 150]},
                {"class": "dog", "confidence": 0.87, "bbox": [120, 50, 200, 180]},
            ]
            
            return detections
        except Exception as e:
            return {"error": str(e)}
    
    def extract_text(self, image_data: str):
        """Extract text from image (OCR)"""
        try:
            image = Image.open(BytesIO(base64.b64decode(image_data)))
            
            # Simulated OCR
            text = "Extracted text from image"
            confidence = 0.92
            
            return {"text": text, "confidence": confidence}
        except Exception as e:
            return {"error": str(e)}
    
    def classify_image(self, image_data: str):
        """Classify image into categories"""
        try:
            image = Image.open(BytesIO(base64.b64decode(image_data)))
            
            classifications = [
                {"category": "outdoor", "score": 0.85},
                {"category": "nature", "score": 0.72},
                {"category": "landscape", "score": 0.68}
            ]
            
            return classifications
        except Exception as e:
            return {"error": str(e)}

analyzer = ImageAnalyzer()

@server.tool(name="detect_objects_in_image")
def detect_objects_in_image(image_base64: str) -> str:
    """Detect objects in an image"""
    try:
        detections = analyzer.detect_objects(image_base64)
        return TextContent(text=json.dumps(detections))
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

@server.tool(name="extract_text_from_image")
def extract_text_from_image(image_base64: str) -> str:
    """Extract text from image using OCR"""
    try:
        result = analyzer.extract_text(image_base64)
        return TextContent(text=json.dumps(result))
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

@server.tool(name="classify_image")
def classify_image(image_base64: str) -> str:
    """Classify image by content"""
    try:
        classifications = analyzer.classify_image(image_base64)
        return TextContent(text=json.dumps(classifications))
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

Time Series Forecasting Backend

from mcp.server import Server
from mcp.types import TextContent
import json
import numpy as np
from datetime import datetime, timedelta

server = Server("forecasting-backend")

class TimeSeriesForecaster:
    def __init__(self):
        # Simulated trained model
        self.model = None
    
    def forecast(self, data_points: list, periods: int = 12, confidence: float = 0.95):
        """Generate time series forecast"""
        try:
            # Simulated ARIMA-like forecasting
            values = np.array(data_points)
            trend = np.polyfit(range(len(values)), values, 1)
            
            forecast = []
            for i in range(periods):
                future_idx = len(values) + i
                predicted = trend[0] * future_idx + trend[1]
                # Add some noise
                predicted += np.random.normal(0, np.std(values) * 0.1)
                forecast.append({
                    "period": i + 1,
                    "forecast": float(predicted),
                    "lower_bound": float(predicted - 2),
                    "upper_bound": float(predicted + 2)
                })
            
            return forecast
        except Exception as e:
            return {"error": str(e)}
    
    def detect_anomalies(self, data_points: list, sensitivity: float = 2.0):
        """Detect anomalies in time series"""
        values = np.array(data_points)
        mean = np.mean(values)
        std = np.std(values)
        
        anomalies = []
        for i, value in enumerate(values):
            z_score = abs((value - mean) / std)
            if z_score > sensitivity:
                anomalies.append({
                    "index": i,
                    "value": float(value),
                    "anomaly_score": float(z_score)
                })
        
        return anomalies

forecaster = TimeSeriesForecaster()

@server.tool(name="forecast_time_series")
def forecast_time_series(
    data_points: list,
    periods: int = 12,
    confidence: float = 0.95
) -> str:
    """Forecast future values in time series"""
    try:
        forecast = forecaster.forecast(data_points, periods, confidence)
        return TextContent(text=json.dumps(forecast))
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

@server.tool(name="detect_anomalies")
def detect_anomalies(data_points: list, sensitivity: float = 2.0) -> str:
    """Detect anomalies in time series data"""
    try:
        anomalies = forecaster.detect_anomalies(data_points, sensitivity)
        return TextContent(text=json.dumps(anomalies))
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

Recommendation Engine Backend

from mcp.server import Server
from mcp.types import TextContent
import json

server = Server("recommendation-backend")

class RecommendationEngine:
    def __init__(self):
        # Simulated user-item interaction data
        self.user_profiles = {
            "user_1": {"interests": ["tech", "sports"], "avg_rating": 4.2},
            "user_2": {"interests": ["music", "art"], "avg_rating": 3.8}
        }
        
        self.items = {
            "item_1": {"category": "tech", "avg_rating": 4.5, "tags": ["AI", "Python"]},
            "item_2": {"category": "sports", "avg_rating": 4.0, "tags": ["fitness"]},
            "item_3": {"category": "music", "avg_rating": 4.3, "tags": ["jazz", "relaxing"]}
        }
    
    def get_recommendations(self, user_id: str, top_k: int = 5):
        """Get personalized recommendations for user"""
        if user_id not in self.user_profiles:
            return {"error": "User not found"}
        
        user = self.user_profiles[user_id]
        user_interests = set(user["interests"])
        
        recommendations = []
        for item_id, item in self.items.items():
            # Simple collaborative filtering
            category_match = item["category"] in user_interests
            
            score = item["avg_rating"]
            if category_match:
                score += 0.5  # Boost score for category match
            
            recommendations.append({
                "item_id": item_id,
                "category": item["category"],
                "rating": item["avg_rating"],
                "match_score": float(score),
                "reason": "Matches your interests" if category_match else "Popular item"
            })
        
        # Sort and return top K
        recommendations.sort(key=lambda x: x["match_score"], reverse=True)
        return recommendations[:top_k]
    
    def predict_user_rating(self, user_id: str, item_id: str):
        """Predict user's rating for an item"""
        if user_id not in self.user_profiles or item_id not in self.items:
            return {"error": "User or item not found"}
        
        user = self.user_profiles[user_id]
        item = self.items[item_id]
        
        # Simple prediction: average of user's and item's ratings
        predicted_rating = (user["avg_rating"] + item["avg_rating"]) / 2
        
        return {
            "predicted_rating": float(predicted_rating),
            "confidence": 0.75,
            "explanation": "Based on your preferences and item popularity"
        }

engine = RecommendationEngine()

@server.tool(name="get_recommendations")
def get_recommendations(user_id: str, top_k: int = 5) -> str:
    """Get personalized recommendations"""
    try:
        recs = engine.get_recommendations(user_id, top_k)
        return TextContent(text=json.dumps(recs))
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

@server.tool(name="predict_user_rating")
def predict_user_rating(user_id: str, item_id: str) -> str:
    """Predict user's rating for item"""
    try:
        prediction = engine.predict_user_rating(user_id, item_id)
        return TextContent(text=json.dumps(prediction))
    except Exception as e:
        return TextContent(text=f"Error: {str(e)}")

Client Interaction

from mcp.client import Client
import asyncio
import json
import base64

async def use_custom_backend():
    # Connect to custom AI backend
    client = Client("domain-llm-backend", transport="stdio")
    
    # Use specialized legal tool
    result = await client.call_tool(
        name="query_legal_documents",
        arguments={
            "question": "What is the refund policy?",
            "document_id": "policy-201"
        }
    )
    
    answer = json.loads(result.content[0].text)
    print(f"Legal Answer: {answer['answer']}")
    print(f"Confidence: {answer['confidence']:.2%}")

async def use_vision_backend():
    # Connect to vision service
    client = Client("vision-backend", transport="stdio")
    
    # Load and encode image
    with open("image.jpg", "rb") as f:
        image_data = base64.b64encode(f.read()).decode()
    
    # Detect objects
    result = await client.call_tool(
        name="detect_objects_in_image",
        arguments={"image_base64": image_data}
    )
    
    detections = json.loads(result.content[0].text)
    for detection in detections:
        print(f"Found {detection['class']}: {detection['confidence']:.1%}")

asyncio.run(use_custom_backend())

Best Practices

• Define tool schemas clearly to help clients understand capabilities
• Implement response caching for expensive computations
• Add input validation and rate limiting
• Provide fallback responses for model failures
• Monitor and log tool usage patterns
• Version your backends for compatibility
• Implement graceful degradation for unavailable services
• Document expected latencies and resource requirements