"""
MCP Tool implementations
Provides executable functions for random number generation and data manipulation
"""
import random
import csv
import statistics
from pathlib import Path
from typing import List, Optional
from datetime import datetime
from mcp.server.fastmcp import FastMCP, Context
from models import (
RandomNumberResult,
SaveResult,
StatisticsResult,
ErrorResponse
)
def register_tools(mcp: FastMCP):
"""Register all tool implementations with the MCP server"""
@mcp.tool()
async def generate_random_numbers(
ctx: Context,
count: int = 5,
min_value: int = 0,
max_value: int = 999,
seed: Optional[int] = None
) -> RandomNumberResult:
"""
Generate random numbers with specified parameters.
Args:
count: Number of random values to generate (1-10000, default: 5)
min_value: Minimum value in range (default: 0)
max_value: Maximum value in range (default: 999)
seed: Random seed for reproducibility (optional)
Returns:
RandomNumberResult containing generated numbers and metadata
Raises:
ValueError: If parameters are invalid
"""
# Log operation
await ctx.info(
f"Generating {count} random numbers in range [{min_value}, {max_value}]"
)
# Validate inputs
if count <= 0 or count > 10000:
error_msg = "Count must be between 1 and 10000"
await ctx.error(error_msg)
raise ValueError(error_msg)
if min_value >= max_value:
error_msg = "min_value must be less than max_value"
await ctx.error(error_msg)
raise ValueError(error_msg)
# Set seed if provided
if seed is not None:
random.seed(seed)
await ctx.debug(f"Using random seed: {seed}")
# Generate random numbers
numbers = [random.randint(min_value, max_value) for _ in range(count)]
# Store in context for resource access
app_ctx = ctx.request_context.lifespan_context
app_ctx.current_data = numbers
# Create result
result = RandomNumberResult(
numbers=numbers,
count=count,
min_value=min_value,
max_value=max_value,
timestamp=datetime.now().isoformat(),
seed=seed
)
await ctx.info(f"Successfully generated {count} random numbers")
return result
@mcp.tool()
async def save_random_data(
ctx: Context,
numbers: List[int],
filename: str = "data.csv"
) -> SaveResult:
"""
Save random numbers to CSV file.
Args:
numbers: List of integers to save
filename: Output filename (default: "data.csv")
Returns:
SaveResult with file information and status
Raises:
IOError: If file write fails
"""
await ctx.info(f"Saving {len(numbers)} numbers to {filename}")
try:
# Get data directory from context
app_ctx = ctx.request_context.lifespan_context
data_dir = app_ctx.data_dir
# Validate filename (security check)
if ".." in filename or "/" in filename or "\\" in filename:
raise ValueError("Invalid filename: path traversal not allowed")
# Ensure .csv extension
if not filename.endswith('.csv'):
filename += '.csv'
# Create full path
filepath = data_dir / filename
# Write to CSV
with open(filepath, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(['data'])
for value in numbers:
writer.writerow([value])
await ctx.info(f"Successfully saved to {filepath}")
return SaveResult(
success=True,
filepath=str(filepath.absolute()),
count=len(numbers),
timestamp=datetime.now().isoformat()
)
except Exception as e:
error_msg = f"Failed to save data: {str(e)}"
await ctx.error(error_msg)
return SaveResult(
success=False,
filepath="",
count=0,
timestamp=datetime.now().isoformat(),
error=error_msg
)
@mcp.tool()
async def analyze_random_data(
ctx: Context,
numbers: List[int]
) -> StatisticsResult:
"""
Analyze statistical properties of random numbers.
Args:
numbers: List of integers to analyze
Returns:
StatisticsResult with computed statistics
Raises:
ValueError: If numbers list is empty
"""
await ctx.info(f"Analyzing {len(numbers)} data points")
if not numbers:
error_msg = "Cannot analyze empty data"
await ctx.error(error_msg)
raise ValueError(error_msg)
# Compute statistics
mean_val = statistics.mean(numbers)
median_val = statistics.median(numbers)
# Standard deviation (handle single-value case)
if len(numbers) > 1:
std_dev_val = statistics.stdev(numbers)
variance_val = statistics.variance(numbers)
else:
std_dev_val = 0.0
variance_val = 0.0
min_val = min(numbers)
max_val = max(numbers)
sum_val = sum(numbers)
range_val = max_val - min_val
result = StatisticsResult(
mean=mean_val,
median=median_val,
std_dev=std_dev_val,
variance=variance_val,
min=min_val,
max=max_val,
count=len(numbers),
sum=sum_val,
range=range_val
)
await ctx.info("Statistical analysis completed")
return result
@mcp.tool()
async def load_data_from_csv(
ctx: Context,
filename: str
) -> List[int]:
"""
Load random numbers from CSV file.
Args:
filename: Name of CSV file to load
Returns:
List of integers loaded from file
Raises:
FileNotFoundError: If file doesn't exist
ValueError: If file format is invalid
"""
await ctx.info(f"Loading data from {filename}")
try:
# Get data directory from context
app_ctx = ctx.request_context.lifespan_context
data_dir = app_ctx.data_dir
# Validate filename
if ".." in filename or "/" in filename or "\\" in filename:
raise ValueError("Invalid filename: path traversal not allowed")
# Create full path
filepath = data_dir / filename
if not filepath.exists():
raise FileNotFoundError(f"File not found: {filename}")
# Read from CSV
data = []
with open(filepath, 'r', encoding='utf-8') as f:
reader = csv.reader(f)
next(reader) # Skip header
for row in reader:
if row:
data.append(int(row[0]))
await ctx.info(f"Loaded {len(data)} values from {filename}")
return data
except Exception as e:
error_msg = f"Failed to load data: {str(e)}"
await ctx.error(error_msg)
raise
@mcp.tool()
async def write_function_notation(
ctx: Context,
filename: str = "func.txt"
) -> str:
"""
Write FGDB function notation to file.
Args:
filename: Output filename (default: "func.txt")
Returns:
Success message with file path
Raises:
IOError: If file write fails
"""
await ctx.info(f"Writing function notation to {filename}")
try:
# Get data directory from context
app_ctx = ctx.request_context.lifespan_context
data_dir = app_ctx.data_dir
# Validate filename
if ".." in filename or "/" in filename or "\\" in filename:
raise ValueError("Invalid filename: path traversal not allowed")
# Create full path
filepath = data_dir / filename
# Write function notation
with open(filepath, 'w', encoding='utf-8') as f:
f.write('y=f0()')
message = f"Function notation written to {filepath.absolute()}"
await ctx.info(message)
return message
except Exception as e:
error_msg = f"Failed to write function notation: {str(e)}"
await ctx.error(error_msg)
raise IOError(error_msg)