MCP Server Implementation Guide

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Any, Optional
import opentelemetry.trace as trace
from azure.ai.ml import MLClient

app = FastAPI(title="Itinerary Planning MCP Server")
tracer = trace.get_tracer("itinerary-planning-server")

@app.post("/mcp/call")
async def call_tool(request: ToolCallRequest) -> ToolCallResponse:
    with tracer.start_as_current_span(f"tool_call_{request.name}") as span:
        try:
            result = await route_tool_call(request.name, request.arguments)
            span.set_status(trace.Status(trace.StatusCode.OK))
            return ToolCallResponse(content=result)
        except Exception as e:
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            raise HTTPException(status_code=400, detail=str(e))

async def route_tool_call(tool_name: str, args: Dict[str, Any]) -> Any:
    match tool_name:
        case "plan_itinerary":
            return await plan_itinerary(args)
        case "optimize_route":
            return await optimize_route(args)
        case "schedule_activities":
            return await schedule_activities(args)
        case "estimate_costs":
            return await estimate_costs(args)
        case _:
            raise ValueError(f"Unknown tool: {tool_name}")

class ItineraryPlanningService:
    def __init__(self):
        self.ai_client = AzureOpenAIClient()
        self.maps_client = AzureMapsClient()
        self.cost_estimator = CostEstimator()
    
    async def plan_itinerary(self, args: Dict[str, Any]) -> ItineraryPlan:
        destinations = args["destinations"]
        preferences = args["preferences"]
        duration = args["duration"]
        
        # Phase 1: Generate base itinerary structure
        with tracer.start_as_current_span("generate_base_structure"):
            base_structure = await self._generate_base_structure(
                destinations, duration, preferences
            )
        
        # Phase 2: Optimize for travel efficiency
        with tracer.start_as_current_span("optimize_travel"):
            optimized_route = await self._optimize_travel_route(
                base_structure, preferences
            )
        
        # Phase 3: Schedule detailed activities
        with tracer.start_as_current_span("schedule_activities"):
            detailed_schedule = await self._schedule_activities(
                optimized_route, preferences
            )
        
        # Phase 4: Add logistics and recommendations
        with tracer.start_as_current_span("add_logistics"):
            final_itinerary = await self._add_logistics(
                detailed_schedule, preferences
            )
        
        return final_itinerary
    
    async def _generate_base_structure(
        self, 
        destinations: List[Dict], 
        duration: int, 
        preferences: Dict
    ) -> BaseItinerary:
        prompt = f"""
        Create a {duration}-day travel itinerary for these destinations:
        {json.dumps(destinations, indent=2)}
        
        Travel preferences:
        {json.dumps(preferences, indent=2)}
        
        Return a JSON structure with:
        - Daily themes and focus areas
        - Major destinations per day
        - Travel time between locations
        - Recommended time allocation
        
        Consider:
        - Logical geographic flow
        - Activity intensity variation
        - Weather and seasonal factors
        - Local events and festivals
        """
        
        response = await self.ai_client.chat_completion(
            messages=[{"role": "user", "content": prompt}],
            temperature=0.7,
            max_tokens=2000
        )
        
        return BaseItinerary.parse_from_ai_response(response)
    
    async def _optimize_travel_route(
        self, 
        base: BaseItinerary, 
        preferences: Dict
    ) -> OptimizedRoute:
        # Use Azure Maps for route optimization
        locations = base.extract_locations()
        
        route_matrix = await self.maps_client.calculate_route_matrix(
            origins=locations,
            destinations=locations,
            travel_mode=preferences.get("transport_mode", "driving")
        )
        
        # Apply traveling salesman optimization
        optimized_order = self._optimize_location_order(
            locations, route_matrix, base.daily_themes
        )
        
        return OptimizedRoute(
            locations=optimized_order,
            travel_times=route_matrix,
            total_distance=sum(route_matrix.distances),
            recommendations=self._generate_route_recommendations(optimized_order)
        )

from pydantic import BaseModel
from typing import List, Dict, Optional
from datetime import datetime, time

class ItineraryPlan(BaseModel):
    id: str
    title: str
    duration_days: int
    destinations: List[str]
    daily_plans: List[DailyPlan]
    cost_estimate: CostEstimate
    metadata: Dict[str, Any]

class DailyPlan(BaseModel):
    day: int
    date: Optional[datetime]
    theme: str
    location: str
    activities: List[Activity]
    meals: List[MealRecommendation]
    accommodation: Optional[AccommodationRecommendation]
    transportation: List[TransportationSegment]
    estimated_cost: float
    notes: List[str]

class Activity(BaseModel):
    name: str
    description: str
    location: str
    start_time: time
    duration_minutes: int
    cost: float
    booking_required: bool
    difficulty_level: str
    weather_dependent: bool
    alternatives: List[str]

class CostEstimate(BaseModel):
    accommodation: float
    transportation: float
    activities: float
    meals: float
    miscellaneous: float
    total: float
    currency: str
    confidence_level: float

from fastapi import FastAPI, HTTPException
import ast
import sys
import io
import contextlib
import traceback
from typing import Dict, Any
import opentelemetry.trace as trace

app = FastAPI(title="Code Evaluation MCP Server")
tracer = trace.get_tracer("code-evaluation-server")

class SafeCodeExecutor:
    ALLOWED_MODULES = {
        'math', 'datetime', 'json', 'statistics', 
        'itertools', 'functools', 'collections'
    }
    
    FORBIDDEN_NAMES = {
        'exec', 'eval', 'compile', 'open', 'file',
        'input', 'raw_input', '__import__', 'reload',
        'vars', 'locals', 'globals', 'dir'
    }
    
    def __init__(self):
        self.global_namespace = self._create_safe_namespace()
    
    def _create_safe_namespace(self) -> Dict[str, Any]:
        # Create restricted namespace with safe built-ins
        safe_builtins = {
            'abs', 'all', 'any', 'bool', 'dict', 'enumerate',
            'filter', 'float', 'int', 'len', 'list', 'map',
            'max', 'min', 'range', 'round', 'sorted', 'str',
            'sum', 'tuple', 'zip'
        }
        
        namespace = {name: getattr(__builtins__, name) 
                    for name in safe_builtins if hasattr(__builtins__, name)}
        
        # Add safe modules
        for module_name in self.ALLOWED_MODULES:
            try:
                namespace[module_name] = __import__(module_name)
            except ImportError:
                pass
        
        return namespace
    
    def execute_code(self, code: str, inputs: Dict[str, Any] = None) -> Dict[str, Any]:
        # Validate code safety
        self._validate_code_safety(code)
        
        # Create execution namespace
        exec_namespace = self.global_namespace.copy()
        if inputs:
            exec_namespace.update(inputs)
        
        # Capture output
        output_buffer = io.StringIO()
        result = None
        
        try:
            with contextlib.redirect_stdout(output_buffer):
                # Execute code
                compiled_code = compile(code, '<string>', 'exec')
                exec(compiled_code, exec_namespace)
                
                # Extract result if available
                if 'result' in exec_namespace:
                    result = exec_namespace['result']
            
            return {
                'success': True,
                'result': result,
                'output': output_buffer.getvalue(),
                'namespace_vars': {k: v for k, v in exec_namespace.items() 
                                 if not k.startswith('_') and k not in self.global_namespace}
            }
            
        except Exception as e:
            return {
                'success': False,
                'error': str(e),
                'error_type': type(e).__name__,
                'traceback': traceback.format_exc()
            }
    
    def _validate_code_safety(self, code: str) -> None:
        # Parse AST and check for forbidden operations
        try:
            tree = ast.parse(code)
        except SyntaxError as e:
            raise ValueError(f"Syntax error in code: {e}")
        
        for node in ast.walk(tree):
            if isinstance(node, ast.Name) and node.id in self.FORBIDDEN_NAMES:
                raise ValueError(f"Forbidden operation: {node.id}")
            elif isinstance(node, ast.Import):
                for alias in node.names:
                    if alias.name not in self.ALLOWED_MODULES:
                        raise ValueError(f"Module not allowed: {alias.name}")

@app.post("/mcp/call")
async def call_tool(request: ToolCallRequest) -> ToolCallResponse:
    with tracer.start_as_current_span(f"tool_call_{request.name}") as span:
        try:
            result = await route_tool_call(request.name, request.arguments)
            span.set_status(trace.Status(trace.StatusCode.OK))
            return ToolCallResponse(content=result)
        except Exception as e:
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
            raise HTTPException(status_code=400, detail=str(e))

async def route_tool_call(tool_name: str, args: Dict[str, Any]) -> Any:
    executor = SafeCodeExecutor()
    
    match tool_name:
        case "execute_python":
            return executor.execute_code(
                args["code"], 
                args.get("inputs", {})
            )
        
        case "calculate_travel_metrics":
            return await calculate_travel_metrics(
                args["calculation_type"],
                args["data"]
            )
        
        case "data_analysis":
            return await perform_data_analysis(
                args["data"],
                args["analysis_type"]
            )
        
        case _:
            raise ValueError(f"Unknown tool: {tool_name}")

async def calculate_travel_metrics(calc_type: str, data: Dict[str, Any]) -> Dict[str, Any]:
    """Calculate travel-specific metrics using safe code execution."""
    
    calculation_templates = {
        "cost_per_day": """
import statistics
costs = data['daily_costs']
result = {
    'average': statistics.mean(costs),
    'median': statistics.median(costs),
    'total': sum(costs),
    'min': min(costs),
    'max': max(costs)
}
""",
        "travel_efficiency": """
import math
distances = data['distances']
times = data['travel_times']
efficiency_scores = [d/t if t > 0 else 0 for d, t in zip(distances, times)]
result = {
    'scores': efficiency_scores,
    'average_efficiency': sum(efficiency_scores) / len(efficiency_scores),
    'total_distance': sum(distances),
    'total_time': sum(times)
}
""",
        "budget_optimization": """
import itertools
activities = data['activities']
budget = data['budget']

# Find optimal activity combination within budget
combinations = []
for r in range(1, len(activities) + 1):
    for combo in itertools.combinations(activities, r):
        total_cost = sum(activity['cost'] for activity in combo)
        total_value = sum(activity['value_score'] for activity in combo)
        if total_cost <= budget:
            combinations.append({
                'activities': [a['name'] for a in combo],
                'cost': total_cost,
                'value': total_value,
                'efficiency': total_value / total_cost if total_cost > 0 else 0
            })

result = sorted(combinations, key=lambda x: x['efficiency'], reverse=True)[:5]
"""
    }
    
    if calc_type not in calculation_templates:
        raise ValueError(f"Unknown calculation type: {calc_type}")
    
    executor = SafeCodeExecutor()
    return executor.execute_code(
        calculation_templates[calc_type],
        {"data": data}
    )

from fastapi import FastAPI, HTTPException
import torch
import onnxruntime as ort
from transformers import AutoTokenizer
from typing import Dict, Any, List
import opentelemetry.trace as trace

app = FastAPI(title="Model Inference MCP Server")
tracer = trace.get_tracer("model-inference-server")

class ModelInferenceService:
    def __init__(self):
        self.models = {}
        self.tokenizers = {}
        self.load_models()
    
    def load_models(self):
        """Load pre-configured models for travel-specific tasks."""
        
        # Load travel classification model
        self.models['travel_classifier'] = ort.InferenceSession(
            "models/travel_classifier.onnx",
            providers=['CUDAExecutionProvider', 'CPUExecutionProvider']
        )
        self.tokenizers['travel_classifier'] = AutoTokenizer.from_pretrained(
            "travel-classifier-tokenizer"
        )
        
        # Load sentiment analysis model
        self.models['sentiment'] = ort.InferenceSession(
            "models/sentiment_analysis.onnx",
            providers=['CUDAExecutionProvider', 'CPUExecutionProvider']
        )
        self.tokenizers['sentiment'] = AutoTokenizer.from_pretrained(
            "sentiment-tokenizer"
        )
    
    async def classify_travel_query(self, text: str) -> Dict[str, Any]:
        """Classify travel query into categories."""
        with tracer.start_as_current_span("classify_travel_query"):
            tokenizer = self.tokenizers['travel_classifier']
            model = self.models['travel_classifier']
            
            # Tokenize input
            inputs = tokenizer(
                text,
                return_tensors="np",
                padding=True,
                truncation=True,
                max_length=512
            )
            
            # Run inference
            outputs = model.run(None, {
                'input_ids': inputs['input_ids'],
                'attention_mask': inputs['attention_mask']
            })
            
            # Process results
            logits = outputs[0]
            probabilities = torch.softmax(torch.from_numpy(logits), dim=-1)
            
            categories = [
                'destination_inquiry', 'accommodation_search', 
                'activity_planning', 'transportation', 'budget_planning'
            ]
            
            results = {
                'categories': {
                    cat: float(prob) 
                    for cat, prob in zip(categories, probabilities[0])
                },
                'primary_category': categories[probabilities.argmax()],
                'confidence': float(probabilities.max())
            }
            
            return results
    
    async def analyze_sentiment(self, text: str) -> Dict[str, Any]:
        """Analyze sentiment of travel-related text."""
        with tracer.start_as_current_span("analyze_sentiment"):
            tokenizer = self.tokenizers['sentiment']
            model = self.models['sentiment']
            
            inputs = tokenizer(
                text,
                return_tensors="np",
                padding=True,
                truncation=True,
                max_length=512
            )
            
            outputs = model.run(None, {
                'input_ids': inputs['input_ids'],
                'attention_mask': inputs['attention_mask']
            })
            
            logits = outputs[0]
            probabilities = torch.softmax(torch.from_numpy(logits), dim=-1)
            
            sentiment_labels = ['negative', 'neutral', 'positive']
            
            return {
                'sentiment': sentiment_labels[probabilities.argmax()],
                'confidence': float(probabilities.max()),
                'scores': {
                    label: float(score)
                    for label, score in zip(sentiment_labels, probabilities[0])
                }
            }