"""
Construction Cost Calculator MCP Server
A no-auth MCP server that reads construction cost data from a public Google Sheet
and provides tools for searching, filtering, and calculating costs.
"""
import os
import time
from typing import Optional
from io import StringIO
import httpx
import pandas as pd
from fastmcp import FastMCP
# Configuration
GOOGLE_SHEET_ID = os.getenv(
"GOOGLE_SHEET_ID",
"1laH5l1FjwWC5HXLp8xkG1eolwScKHIlxB6PohBWQjh0"
)
SHEET_GID = os.getenv("SHEET_GID", "0")
CACHE_TTL_SECONDS = int(os.getenv("CACHE_TTL_SECONDS", "300")) # 5 minutes default
DEFAULT_LABOR_RATE = float(os.getenv("DEFAULT_LABOR_RATE", "75.0")) # $/hour
# Cache for sheet data
_cache = {
"data": None,
"last_fetch": 0
}
# Initialize FastMCP server
mcp = FastMCP("Construction Cost Calculator")
def get_csv_url() -> str:
"""Get the CSV export URL for the Google Sheet."""
return f"https://docs.google.com/spreadsheets/d/{GOOGLE_SHEET_ID}/export?format=csv&gid={SHEET_GID}"
async def fetch_sheet_data() -> pd.DataFrame:
"""
Fetch data from Google Sheet with caching.
Returns cached data if still valid, otherwise fetches fresh data.
"""
current_time = time.time()
# Check if cache is still valid
if _cache["data"] is not None and (current_time - _cache["last_fetch"]) < CACHE_TTL_SECONDS:
return _cache["data"]
# Fetch fresh data
async with httpx.AsyncClient() as client:
response = await client.get(get_csv_url(), follow_redirects=True)
response.raise_for_status()
# Parse CSV
df = pd.read_csv(StringIO(response.text))
# Clean column names (remove any whitespace)
df.columns = df.columns.str.strip()
# Update cache
_cache["data"] = df
_cache["last_fetch"] = current_time
return df
def format_item(row: pd.Series) -> dict:
"""Format a single item row as a dictionary."""
return {
"item_code": row["item_code"],
"description": row["description"],
"category": row["category"],
"material_cost": float(row["material_cost"]),
"labor_hours": float(row["labor_hours"]),
"unit": row["unit"]
}
@mcp.tool()
async def list_all_items() -> dict:
"""
List all available construction items from the cost database.
Returns a list of all items with their codes, descriptions, categories,
material costs, labor hours, and units.
"""
df = await fetch_sheet_data()
items = [format_item(row) for _, row in df.iterrows()]
categories = df["category"].unique().tolist()
return {
"total_items": len(items),
"categories": categories,
"items": items
}
@mcp.tool()
async def get_item(item_code: str) -> dict:
"""
Get detailed information for a specific construction item by its code.
Args:
item_code: The unique code for the item (e.g., 'concrete_slab_4000psi')
Returns the item details including material cost, labor hours, and unit.
"""
df = await fetch_sheet_data()
# Find the item
item_row = df[df["item_code"] == item_code]
if item_row.empty:
# Try case-insensitive search
item_row = df[df["item_code"].str.lower() == item_code.lower()]
if item_row.empty:
available_codes = df["item_code"].tolist()
return {
"error": f"Item '{item_code}' not found",
"suggestion": "Use list_all_items or search_items to find valid item codes",
"available_codes_sample": available_codes[:10]
}
return {
"found": True,
"item": format_item(item_row.iloc[0])
}
@mcp.tool()
async def search_items(query: str) -> dict:
"""
Search for construction items by keyword in their code or description.
Args:
query: Search term to find in item codes or descriptions
Returns matching items sorted by relevance.
"""
df = await fetch_sheet_data()
query_lower = query.lower()
# Search in both item_code and description
mask = (
df["item_code"].str.lower().str.contains(query_lower, na=False) |
df["description"].str.lower().str.contains(query_lower, na=False)
)
matches = df[mask]
items = [format_item(row) for _, row in matches.iterrows()]
return {
"query": query,
"total_matches": len(items),
"items": items
}
@mcp.tool()
async def get_items_by_category(category: str) -> dict:
"""
Get all construction items in a specific category.
Args:
category: The category name (e.g., 'concrete', 'framing', 'finishes',
'electrical', 'plumbing', 'hvac', 'roofing', 'exterior')
Returns all items in the specified category.
"""
df = await fetch_sheet_data()
category_lower = category.lower()
# Filter by category (case-insensitive)
matches = df[df["category"].str.lower() == category_lower]
if matches.empty:
available_categories = df["category"].unique().tolist()
return {
"error": f"Category '{category}' not found",
"available_categories": available_categories
}
items = [format_item(row) for _, row in matches.iterrows()]
return {
"category": category,
"total_items": len(items),
"items": items
}
@mcp.tool()
async def calculate_cost(
items: list[dict],
labor_rate: Optional[float] = None
) -> dict:
"""
Calculate the total cost for a list of construction items with quantities.
Args:
items: List of items with 'item_code' and 'quantity' keys.
Example: [{"item_code": "concrete_slab_4000psi", "quantity": 1000}]
labor_rate: Hourly labor rate in dollars (default: $75/hour)
Returns itemized costs and total project cost.
Cost Formula:
Item Cost = (Material Cost × Quantity) + (Labor Hours × Quantity × Labor Rate)
"""
df = await fetch_sheet_data()
if labor_rate is None:
labor_rate = DEFAULT_LABOR_RATE
line_items = []
total_material_cost = 0.0
total_labor_cost = 0.0
total_labor_hours = 0.0
errors = []
for item_request in items:
item_code = item_request.get("item_code")
quantity = item_request.get("quantity", 0)
if not item_code:
errors.append({"error": "Missing item_code in request"})
continue
# Find the item
item_row = df[df["item_code"] == item_code]
if item_row.empty:
# Try case-insensitive
item_row = df[df["item_code"].str.lower() == item_code.lower()]
if item_row.empty:
errors.append({
"item_code": item_code,
"error": "Item not found"
})
continue
item_data = item_row.iloc[0]
material_cost = float(item_data["material_cost"])
labor_hours = float(item_data["labor_hours"])
# Calculate costs
item_material_cost = material_cost * quantity
item_labor_hours = labor_hours * quantity
item_labor_cost = item_labor_hours * labor_rate
item_total = item_material_cost + item_labor_cost
line_items.append({
"item_code": item_code,
"description": item_data["description"],
"quantity": quantity,
"unit": item_data["unit"],
"material_cost_per_unit": material_cost,
"labor_hours_per_unit": labor_hours,
"material_cost": round(item_material_cost, 2),
"labor_hours": round(item_labor_hours, 2),
"labor_cost": round(item_labor_cost, 2),
"total_cost": round(item_total, 2)
})
total_material_cost += item_material_cost
total_labor_cost += item_labor_cost
total_labor_hours += item_labor_hours
grand_total = total_material_cost + total_labor_cost
return {
"labor_rate_used": labor_rate,
"line_items": line_items,
"summary": {
"total_material_cost": round(total_material_cost, 2),
"total_labor_hours": round(total_labor_hours, 2),
"total_labor_cost": round(total_labor_cost, 2),
"grand_total": round(grand_total, 2)
},
"errors": errors if errors else None
}
# Run the server
if __name__ == "__main__":
port = int(os.getenv("PORT", "8000"))
mcp.run(transport="streamable-http", host="0.0.0.0", port=port)
