"""
MCP Resource implementations
Provides access to data and metadata through URI-based resources
"""
import json
import csv
from pathlib import Path
from typing import List
from datetime import datetime
from mcp.server.fastmcp import FastMCP, Context
def register_resources(mcp: FastMCP):
"""Register all resource implementations with the MCP server"""
@mcp.resource("random://current")
async def get_current_data(ctx: Context) -> str:
"""
Get the most recently generated random numbers.
Returns:
JSON string containing current data or error message
"""
app_ctx = ctx.request_context.lifespan_context
if app_ctx.current_data is None:
return json.dumps({
"error": "No data generated yet",
"message": "Use generate_random_numbers tool first"
}, indent=2)
return json.dumps({
"numbers": app_ctx.current_data,
"count": len(app_ctx.current_data),
"timestamp": datetime.now().isoformat()
}, indent=2)
@mcp.resource("file://data/{filename}")
async def get_saved_data(ctx: Context, filename: str) -> str:
"""
Read saved random data from CSV file.
Args:
filename: Name of the CSV file to read
Returns:
JSON string containing file data or error message
"""
try:
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")
filepath = data_dir / filename
if not filepath.exists():
return json.dumps({
"error": "File not found",
"filename": filename
}, indent=2)
# Read CSV file
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]))
return json.dumps({
"filename": filename,
"filepath": str(filepath.absolute()),
"numbers": data,
"count": len(data),
"timestamp": datetime.now().isoformat()
}, indent=2)
except Exception as e:
return json.dumps({
"error": str(e),
"filename": filename
}, indent=2)
@mcp.resource("meta://statistics")
async def get_statistics_metadata(ctx: Context) -> str:
"""
Get statistical metadata about all generated data.
Returns:
JSON string containing statistics metadata
"""
app_ctx = ctx.request_context.lifespan_context
if app_ctx.current_data is None:
return json.dumps({
"error": "No data available",
"message": "Generate data first using generate_random_numbers tool"
}, indent=2)
# Compute basic statistics
numbers = app_ctx.current_data
count = len(numbers)
if count > 0:
mean_val = sum(numbers) / count
min_val = min(numbers)
max_val = max(numbers)
# Median calculation
sorted_nums = sorted(numbers)
if count % 2 == 0:
median_val = (sorted_nums[count // 2 - 1] + sorted_nums[count // 2]) / 2
else:
median_val = sorted_nums[count // 2]
else:
mean_val = median_val = min_val = max_val = 0
return json.dumps({
"count": count,
"mean": mean_val,
"median": median_val,
"min": min_val,
"max": max_val,
"range": max_val - min_val if count > 0 else 0,
"timestamp": datetime.now().isoformat()
}, indent=2)
@mcp.resource("meta://files")
async def list_data_files(ctx: Context) -> str:
"""
List all data files in the data directory.
Returns:
JSON string containing list of files with metadata
"""
try:
app_ctx = ctx.request_context.lifespan_context
data_dir = app_ctx.data_dir
files = []
for filepath in data_dir.glob("*.csv"):
stat = filepath.stat()
files.append({
"filename": filepath.name,
"filepath": str(filepath.absolute()),
"size_bytes": stat.st_size,
"created_at": datetime.fromtimestamp(stat.st_ctime).isoformat(),
"modified_at": datetime.fromtimestamp(stat.st_mtime).isoformat()
})
return json.dumps({
"data_directory": str(data_dir.absolute()),
"file_count": len(files),
"files": files,
"timestamp": datetime.now().isoformat()
}, indent=2)
except Exception as e:
return json.dumps({
"error": str(e)
}, indent=2)
@mcp.resource("meta://server-info")
async def get_server_info(ctx: Context) -> str:
"""
Get server information and capabilities.
Returns:
JSON string containing server metadata
"""
app_ctx = ctx.request_context.lifespan_context
return json.dumps({
"name": "f0_random_server",
"version": "1.0.0",
"description": "FGDB f0_make_randomvalues function block MCP server",
"capabilities": {
"tools": [
"generate_random_numbers",
"save_random_data",
"analyze_random_data",
"load_data_from_csv",
"write_function_notation"
],
"resources": [
"random://current",
"file://data/{filename}",
"meta://statistics",
"meta://files",
"meta://server-info"
],
"prompts": [
"generate-test-dataset",
"analyze-data"
]
},
"data_directory": str(app_ctx.data_dir.absolute()),
"has_current_data": app_ctx.current_data is not None,
"timestamp": datetime.now().isoformat()
}, indent=2)