"""
Extended MCP Server for M/M/1 and M/M/c Queue Simulation
Now supports both single server (M/M/1) and multiple servers (M/M/c)
"""
try:
from fastmcp import FastMCP
except ImportError:
from mcp.server.fastmcp import FastMCP
from typing import Dict, Any, Optional
import json
# Import M/M/c simulation
from .simulations.mmc_queue import (
MMcConfig,
MMcQueueSimulation,
calculate_mmc_theoretical
)
# Initialize FastMCP Server
mcp = FastMCP("MM1/MMc Simulation Server", version="0.2.0")
# ============================================================================
# TOOLS: M/M/c Support
# ============================================================================
@mcp.tool()
async def run_mmc_simulation(
arrival_rate: float,
service_rate: float,
num_servers: int = 1,
simulation_time: float = 10000.0,
random_seed: Optional[int] = None
) -> Dict[str, Any]:
"""
Run M/M/c queue simulation (supports 1 to n servers)
This tool can simulate:
- M/M/1: Single server (num_servers=1)
- M/M/2: Two servers (num_servers=2)
- M/M/3: Three servers (num_servers=3)
- M/M/c: Any number of servers
Args:
arrival_rate: λ (customers per time unit)
service_rate: μ (customers per time unit per server)
num_servers: c (number of parallel servers)
simulation_time: Duration of simulation
random_seed: Random seed (None for random)
Returns:
Simulation and theoretical metrics for M/M/c system
Example:
# Supermarket with 3 cashiers
run_mmc_simulation(
arrival_rate=12, # 12 customers/min arrive
service_rate=5, # Each cashier serves 5/min
num_servers=3 # 3 cashiers
)
"""
import time
# Handle random seed
if random_seed is None:
random_seed = int(time.time() * 1000) % 2147483647
# Validate stability
rho = arrival_rate / (num_servers * service_rate)
if rho >= 1:
return {
"error": "System unstable",
"details": {
"utilization": rho,
"message": f"ρ = λ/(c*μ) = {arrival_rate}/({num_servers}*{service_rate}) = {rho:.3f} >= 1",
"suggestion": f"Need λ < {num_servers * service_rate} or more servers"
}
}
try:
# Create config
config = MMcConfig(
arrival_rate=arrival_rate,
service_rate=service_rate,
num_servers=num_servers,
simulation_time=simulation_time,
random_seed=random_seed
)
# Run simulation
sim = MMcQueueSimulation(config)
sim_results = sim.run()
# Calculate theoretical
theo_results = calculate_mmc_theoretical(
arrival_rate, service_rate, num_servers
)
# Compare results
comparison = {
"utilization_error": abs(sim_results["utilization"] - theo_results["utilization"]) / theo_results["utilization"] * 100,
"waiting_time_error": abs(sim_results["avg_waiting_time"] - theo_results["avg_waiting_time"]) / theo_results["avg_waiting_time"] * 100,
"queue_length_error": abs(sim_results["avg_queue_length"] - theo_results["avg_queue_length"]) / theo_results["avg_queue_length"] * 100
}
return {
"model": f"M/M/{num_servers}",
"configuration": {
"arrival_rate": arrival_rate,
"service_rate": service_rate,
"num_servers": num_servers,
"utilization": rho,
"simulation_time": simulation_time,
"random_seed": random_seed
},
"simulation_results": sim_results,
"theoretical_results": theo_results,
"accuracy": {
"mape": sum(comparison.values()) / len(comparison),
"errors": comparison
}
}
except Exception as e:
return {
"error": str(e),
"type": type(e).__name__
}
@mcp.tool()
async def compare_mm1_vs_mmc(
arrival_rate: float,
service_rate_per_server: float,
num_servers: int = 3,
simulation_time: float = 10000.0
) -> Dict[str, Any]:
"""
Compare M/M/1 × c (separate queues) vs M/M/c (pooled queue)
This is the key tool for comparing:
- Strategy A: c separate M/M/1 queues (distributed)
- Strategy B: Single M/M/c queue (pooled)
Perfect for supermarket/bank/coffee shop scenarios!
Args:
arrival_rate: Total arrival rate λ
service_rate_per_server: Service rate μ per server
num_servers: Number of servers c
simulation_time: Simulation duration
Returns:
Comparison showing why pooled queue is better
Example:
# 3 cashiers: separate vs pooled queues
compare_mm1_vs_mmc(
arrival_rate=12,
service_rate_per_server=5,
num_servers=3
)
"""
import time
import random
seed = int(time.time() * 1000) % 2147483647
# Strategy A: M/M/1 × c (each queue gets λ/c arrivals)
mm1_results = []
for i in range(num_servers):
config = MMcConfig(
arrival_rate=arrival_rate / num_servers, # Split arrivals
service_rate=service_rate_per_server,
num_servers=1,
simulation_time=simulation_time,
random_seed=seed + i
)
sim = MMcQueueSimulation(config)
result = sim.run()
mm1_results.append(result)
# Average across all M/M/1 queues
mm1_avg = {
"avg_waiting_time": sum(r["avg_waiting_time"] for r in mm1_results) / num_servers,
"avg_queue_length": sum(r["avg_queue_length"] for r in mm1_results), # Total
"avg_system_time": sum(r["avg_system_time"] for r in mm1_results) / num_servers,
"utilization": sum(r["utilization"] for r in mm1_results) / num_servers,
"max_queue_seen": max(r["max_queue_length"] for r in mm1_results)
}
# Strategy B: M/M/c (pooled queue)
mmc_config = MMcConfig(
arrival_rate=arrival_rate,
service_rate=service_rate_per_server,
num_servers=num_servers,
simulation_time=simulation_time,
random_seed=seed
)
mmc_sim = MMcQueueSimulation(mmc_config)
mmc_results = mmc_sim.run()
# Theoretical values for validation
mm1_theo = calculate_mmc_theoretical(
arrival_rate / num_servers,
service_rate_per_server,
1
)
mmc_theo = calculate_mmc_theoretical(
arrival_rate,
service_rate_per_server,
num_servers
)
# Calculate improvements
wait_improvement = (mm1_avg["avg_waiting_time"] - mmc_results["avg_waiting_time"]) / mm1_avg["avg_waiting_time"] * 100
queue_improvement = (mm1_avg["avg_queue_length"] - mmc_results["avg_queue_length"]) / mm1_avg["avg_queue_length"] * 100
return {
"scenario": {
"total_arrival_rate": arrival_rate,
"service_rate_per_server": service_rate_per_server,
"num_servers": num_servers,
"system_utilization": arrival_rate / (num_servers * service_rate_per_server)
},
"strategy_a_separate_queues": {
"model": f"M/M/1 × {num_servers}",
"simulation": mm1_avg,
"theoretical": {
"avg_waiting_time": mm1_theo["avg_waiting_time"],
"avg_queue_length": mm1_theo["avg_queue_length"] * num_servers,
"avg_system_time": mm1_theo["avg_system_time"]
}
},
"strategy_b_pooled_queue": {
"model": f"M/M/{num_servers}",
"simulation": {
"avg_waiting_time": mmc_results["avg_waiting_time"],
"avg_queue_length": mmc_results["avg_queue_length"],
"avg_system_time": mmc_results["avg_system_time"],
"utilization": mmc_results["utilization"],
"max_queue_seen": mmc_results["max_queue_length"]
},
"theoretical": {
"avg_waiting_time": mmc_theo["avg_waiting_time"],
"avg_queue_length": mmc_theo["avg_queue_length"],
"avg_system_time": mmc_theo["avg_system_time"]
}
},
"pooled_queue_benefits": {
"waiting_time_reduction": f"{wait_improvement:.1f}%",
"queue_length_reduction": f"{queue_improvement:.1f}%",
"better_load_balancing": "Yes - servers share the workload",
"fairness": "FIFO guaranteed across all customers",
"recommendation": "✅ Use pooled queue (M/M/c)" if wait_improvement > 0 else "Use separate queues"
},
"visualization": {
"separate_queues": f"[Q1: {'█' * int(mm1_results[0]['avg_queue_length'])}] → Server1\n" +
f"[Q2: {'█' * int(mm1_results[1]['avg_queue_length'] if len(mm1_results) > 1 else 0)}] → Server2\n" +
f"[Q3: {'█' * int(mm1_results[2]['avg_queue_length'] if len(mm1_results) > 2 else 0)}] → Server3",
"pooled_queue": f"[Q: {'█' * int(mmc_results['avg_queue_length'])}] → {num_servers} Servers"
}
}
@mcp.tool()
async def analyze_cashier_problem(
num_customers_waiting: int = 100,
arrival_rate: float = 12.0,
service_rate: float = 5.0,
num_cashiers: int = 3
) -> Dict[str, Any]:
"""
Analyze the supermarket cashier problem specifically
Compares separate queues vs single pooled queue for cashiers.
Includes time to clear initial queue of waiting customers.
Args:
num_customers_waiting: Initial queue size
arrival_rate: New arrivals per minute
service_rate: Customers served per minute per cashier
num_cashiers: Number of cashiers
Returns:
Complete analysis with recommendations
"""
# Time to clear initial customers (rough estimate)
clearing_rate_pooled = num_cashiers * service_rate - arrival_rate
clearing_rate_separate = service_rate - (arrival_rate / num_cashiers)
if clearing_rate_pooled <= 0:
return {"error": "System overloaded - cannot clear queue"}
time_to_clear_pooled = num_customers_waiting / clearing_rate_pooled
time_to_clear_separate = num_customers_waiting / (num_cashiers * clearing_rate_separate) if clearing_rate_separate > 0 else float('inf')
# Run comparison
comparison = await compare_mm1_vs_mmc(
arrival_rate=arrival_rate,
service_rate_per_server=service_rate,
num_servers=num_cashiers,
simulation_time=max(time_to_clear_pooled * 2, 1000)
)
return {
"problem": {
"description": f"Supermarket with {num_cashiers} cashiers",
"initial_queue": f"{num_customers_waiting} customers waiting",
"arrival_rate": f"{arrival_rate} customers/minute",
"service_rate": f"{service_rate} customers/minute per cashier"
},
"initial_queue_clearing": {
"pooled_queue": {
"time_to_clear": f"{time_to_clear_pooled:.1f} minutes",
"strategy": "All cashiers work together"
},
"separate_queues": {
"time_to_clear": f"{time_to_clear_separate:.1f} minutes" if time_to_clear_separate != float('inf') else "Cannot clear",
"issue": "Some queues may grow while others empty"
}
},
"steady_state_analysis": comparison,
"recommendation": {
"best_strategy": "Single pooled queue (M/M/c)",
"reasons": [
f"Clears initial queue {(time_to_clear_separate/time_to_clear_pooled - 1)*100:.0f}% faster",
f"Reduces average wait by {comparison['pooled_queue_benefits']['waiting_time_reduction']}",
"Guarantees fairness (FIFO)",
"Better server utilization",
"No queue selection anxiety for customers"
],
"implementation": "Use a single queue with multiple cashiers like banks do"
}
}
# Add prompt for explaining M/M/c
@mcp.prompt()
async def explain_mmc_theory() -> Dict[str, Any]:
"""
Explain M/M/c queuing theory and Erlang C formula
"""
return {
"prompt": """
You are an expert in queuing theory. Explain M/M/c queuing systems:
1. **What is M/M/c?**
- First M: Markovian (Poisson) arrivals with rate λ
- Second M: Markovian (exponential) service with rate μ per server
- c: Number of identical parallel servers
2. **Key Formulas:**
- Utilization: ρ = λ/(c×μ) must be < 1 for stability
- Erlang C formula for P(wait): Complex but computable
- Average wait time: Wq = Pc/(c×μ - λ)
3. **M/M/1 vs M/M/c Comparison:**
- M/M/1 × c: Separate queues, customers choose shortest
- M/M/c: Single queue, next customer goes to next free server
- M/M/c is ALWAYS better (lower wait times, fair, balanced)
4. **Real-world Examples:**
- Banks: Single queue → multiple tellers (M/M/c)
- Supermarkets: Often separate queues (suboptimal)
- Call centers: Automatic distribution (M/M/c)
- Airport security: Single queue → multiple scanners
5. **Why Pooled Queue Wins:**
- Statistical multiplexing gain
- No idle servers while customers wait elsewhere
- Automatic load balancing
- Fairness guaranteed
Provide numerical examples showing the dramatic improvement.
"""
}
if __name__ == "__main__":
# Start server
import asyncio
async def test():
# Test M/M/3 simulation
result = await run_mmc_simulation(
arrival_rate=12,
service_rate=5,
num_servers=3
)
print(json.dumps(result, indent=2))
# For actual MCP server deployment
mcp.run()