zillow_mcp_server.py•27.5 kB
#!/usr/bin/env python3
"""
Zillow MCP Server
A Model Context Protocol (MCP) server that provides real-time access to Zillow real estate data.
"""
import os
import json
import logging
import argparse
import backoff
import requests
from typing import List, Optional, Dict, Any, Union
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP, Context
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("zillow-mcp-server")
# Load environment variables from .env file
load_dotenv()
# Get API key from environment variable
ZILLOW_API_KEY = os.getenv("ZILLOW_API_KEY")
if not ZILLOW_API_KEY:
logger.warning("ZILLOW_API_KEY not found in environment variables. Please set it in .env file.")
# Base URL for Zillow API
ZILLOW_API_BASE_URL = "https://api.bridgeinteractive.com/v2"
# Create MCP server
server = FastMCP("zillow")
class ZillowAPIError(Exception):
"""Exception raised for Zillow API errors."""
pass
@backoff.on_exception(backoff.expo,
(requests.exceptions.RequestException, ZillowAPIError),
max_tries=5)
def zillow_api_request(endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""
Make a request to the Zillow API with automatic retries and error handling.
Args:
endpoint: API endpoint path
params: Query parameters
Returns:
JSON response from the API
"""
if not ZILLOW_API_KEY:
raise ZillowAPIError("Zillow API key not configured")
headers = {
"Authorization": f"Bearer {ZILLOW_API_KEY}",
"Content-Type": "application/json",
"Accept": "application/json"
}
url = f"{ZILLOW_API_BASE_URL}/{endpoint}"
try:
response = requests.get(url, headers=headers, params=params, timeout=10)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
logger.warning("Rate limit exceeded. Backing off...")
raise ZillowAPIError("Rate limit exceeded")
elif response.status_code == 401:
logger.error("Authentication failed. Check your API key.")
raise ZillowAPIError("Authentication failed")
else:
logger.error(f"HTTP error: {e}")
raise ZillowAPIError(f"HTTP error: {e}")
except requests.exceptions.RequestException as e:
logger.error(f"Request error: {e}")
raise
@server.tool()
def search_properties(
location: str,
type: str = "forSale",
min_price: Optional[int] = None,
max_price: Optional[int] = None,
beds_min: Optional[int] = None,
beds_max: Optional[int] = None,
baths_min: Optional[float] = None,
baths_max: Optional[float] = None,
home_types: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Search for properties based on various criteria.
Args:
location: Location to search (city, neighborhood, zip code, etc.)
type: Type of listing ("forSale", "forRent", "sold", "recentlySold")
min_price: Minimum price
max_price: Maximum price
beds_min: Minimum number of bedrooms
beds_max: Maximum number of bedrooms
baths_min: Minimum number of bathrooms
baths_max: Maximum number of bathrooms
home_types: List of home types (e.g., ["house", "apartment", "condo"])
Returns:
List of properties matching the criteria
"""
logger.info(f"Searching for properties in {location}")
params = {
"location": location,
"listingType": type
}
# Add optional parameters if provided
if min_price is not None:
params["minPrice"] = min_price
if max_price is not None:
params["maxPrice"] = max_price
if beds_min is not None:
params["minBeds"] = beds_min
if beds_max is not None:
params["maxBeds"] = beds_max
if baths_min is not None:
params["minBaths"] = baths_min
if baths_max is not None:
params["maxBaths"] = baths_max
if home_types is not None:
params["homeTypes"] = ",".join(home_types)
try:
results = zillow_api_request("properties/search", params)
# Format the results for better readability
properties = results.get("properties", [])
formatted_results = []
for prop in properties:
formatted_results.append({
"id": prop.get("id"),
"address": prop.get("address", {}).get("full"),
"price": prop.get("price"),
"beds": prop.get("beds"),
"baths": prop.get("baths"),
"sqft": prop.get("livingArea"),
"type": prop.get("homeType"),
"url": prop.get("detailUrl")
})
return {
"message": f"Found {len(formatted_results)} properties in {location}",
"properties": formatted_results
}
except ZillowAPIError as e:
return {
"error": f"Error searching properties: {str(e)}"
}
@server.tool()
def get_property_details(
property_id: str = None,
address: str = None
) -> Dict[str, Any]:
"""
Get detailed information about a specific property.
Args:
property_id: Zillow property ID
address: Property address (used if property_id is not provided)
Returns:
Detailed property information
"""
if not property_id and not address:
return {
"error": "Either property_id or address must be provided"
}
logger.info(f"Getting property details for {'ID: ' + property_id if property_id else 'Address: ' + address}")
try:
params = {}
if property_id:
params["propertyId"] = property_id
else:
params["address"] = address
result = zillow_api_request("properties/detail", params)
# Format the property details
property_data = result.get("property", {})
formatted_details = {
"id": property_data.get("id"),
"address": property_data.get("address", {}).get("full"),
"price": property_data.get("price"),
"beds": property_data.get("beds"),
"baths": property_data.get("baths"),
"sqft": property_data.get("livingArea"),
"lot_size": property_data.get("lotSize"),
"year_built": property_data.get("yearBuilt"),
"home_type": property_data.get("homeType"),
"description": property_data.get("description"),
"features": property_data.get("features", []),
"schools": property_data.get("schools", []),
"tax_history": property_data.get("taxHistory", []),
"price_history": property_data.get("priceHistory", []),
"zestimate": property_data.get("zestimate"),
"photos": property_data.get("photos", []),
"url": property_data.get("detailUrl")
}
return {
"message": f"Details for property at {formatted_details['address']}",
"property": formatted_details
}
except ZillowAPIError as e:
return {
"error": f"Error getting property details: {str(e)}"
}
@server.tool()
def get_zestimate(
property_id: str = None,
address: str = None
) -> Dict[str, Any]:
"""
Get Zillow's estimated value for a property.
Args:
property_id: Zillow property ID
address: Property address (used if property_id is not provided)
Returns:
Zestimate information
"""
if not property_id and not address:
return {
"error": "Either property_id or address must be provided"
}
logger.info(f"Getting Zestimate for {'ID: ' + property_id if property_id else 'Address: ' + address}")
try:
params = {}
if property_id:
params["propertyId"] = property_id
else:
params["address"] = address
result = zillow_api_request("properties/zestimate", params)
zestimate_data = result.get("zestimate", {})
formatted_zestimate = {
"property_id": result.get("propertyId"),
"address": result.get("address", {}).get("full"),
"zestimate": zestimate_data.get("value"),
"low_range": zestimate_data.get("lowRange"),
"high_range": zestimate_data.get("highRange"),
"last_updated": zestimate_data.get("lastUpdated"),
"rent_zestimate": zestimate_data.get("rentZestimate", {}).get("value")
}
return {
"message": f"Zestimate for {formatted_zestimate['address']}: ${formatted_zestimate['zestimate']:,}",
"zestimate": formatted_zestimate
}
except ZillowAPIError as e:
return {
"error": f"Error getting Zestimate: {str(e)}"
}
@server.tool()
def get_market_trends(
location: str,
metrics: List[str] = ["median_list_price", "median_sale_price", "median_days_on_market"],
time_period: str = "1year"
) -> Dict[str, Any]:
"""
Get real estate market trends for a specific location.
Args:
location: Location to get trends for (city, neighborhood, zip code, etc.)
metrics: List of metrics to retrieve
time_period: Time period for trends ("1month", "3months", "6months", "1year", "5years", "10years")
Returns:
Market trend data
"""
logger.info(f"Getting market trends for {location}")
try:
params = {
"location": location,
"metrics": ",".join(metrics),
"period": time_period
}
result = zillow_api_request("market/trends", params)
trends_data = result.get("trends", {})
formatted_trends = {
"location": result.get("location"),
"period": result.get("period"),
"trends": {}
}
for metric, values in trends_data.items():
formatted_trends["trends"][metric] = {
"current": values.get("current"),
"historical": values.get("historical", []),
"change_ytd": values.get("changeYTD"),
"change_mom": values.get("changeMoM")
}
return {
"message": f"Market trends for {location} over {time_period}",
"trends": formatted_trends
}
except ZillowAPIError as e:
return {
"error": f"Error getting market trends: {str(e)}"
}
@server.tool()
def calculate_mortgage(
home_price: int,
down_payment: int = None,
down_payment_percent: float = None,
loan_term: int = 30,
interest_rate: float = 6.5,
annual_property_tax: int = None,
annual_homeowners_insurance: int = None,
monthly_hoa: int = 0,
include_pmi: bool = True
) -> Dict[str, Any]:
"""
Calculate mortgage payments and related costs.
Args:
home_price: Home purchase price
down_payment: Down payment amount in dollars
down_payment_percent: Down payment as a percentage of home price
loan_term: Loan term in years
interest_rate: Annual interest rate (percentage)
annual_property_tax: Annual property tax amount
annual_homeowners_insurance: Annual homeowners insurance cost
monthly_hoa: Monthly HOA fees
include_pmi: Whether to include PMI for down payments < 20%
Returns:
Mortgage calculation details
"""
logger.info(f"Calculating mortgage for ${home_price:,} home")
# Calculate down payment if not provided
if down_payment is None and down_payment_percent is None:
down_payment_percent = 20.0
if down_payment is None:
down_payment = int(home_price * (down_payment_percent / 100))
else:
down_payment_percent = (down_payment / home_price) * 100
# Calculate loan amount
loan_amount = home_price - down_payment
# Calculate monthly interest rate
monthly_interest_rate = (interest_rate / 100) / 12
# Calculate number of payments
num_payments = loan_term * 12
# Calculate principal and interest payment
if monthly_interest_rate == 0:
monthly_pi = loan_amount / num_payments
else:
monthly_pi = loan_amount * (monthly_interest_rate * (1 + monthly_interest_rate) ** num_payments) / ((1 + monthly_interest_rate) ** num_payments - 1)
# Calculate PMI (Private Mortgage Insurance)
monthly_pmi = 0
if include_pmi and down_payment_percent < 20:
# Typical PMI is 0.5% to 1% of loan amount annually
pmi_rate = 0.007 # 0.7% annual rate
monthly_pmi = (loan_amount * pmi_rate) / 12
# Calculate property tax and insurance
monthly_property_tax = 0
if annual_property_tax is not None:
monthly_property_tax = annual_property_tax / 12
else:
# Estimate property tax if not provided (national average ~1.1% of home value)
estimated_tax_rate = 0.011
monthly_property_tax = (home_price * estimated_tax_rate) / 12
monthly_insurance = 0
if annual_homeowners_insurance is not None:
monthly_insurance = annual_homeowners_insurance / 12
else:
# Estimate insurance if not provided (national average ~$1,200/year)
estimated_annual_insurance = max(1200, home_price * 0.0035) # 0.35% of home value or $1,200, whichever is higher
monthly_insurance = estimated_annual_insurance / 12
# Calculate total monthly payment
monthly_payment = monthly_pi + monthly_pmi + monthly_property_tax + monthly_insurance + monthly_hoa
# Calculate total payment over loan term
total_payment = monthly_payment * num_payments
# Calculate total interest paid
total_interest = (monthly_pi * num_payments) - loan_amount
result = {
"home_price": home_price,
"down_payment": down_payment,
"down_payment_percent": down_payment_percent,
"loan_amount": loan_amount,
"loan_term_years": loan_term,
"interest_rate": interest_rate,
"monthly_payment": {
"principal_and_interest": round(monthly_pi, 2),
"property_tax": round(monthly_property_tax, 2),
"homeowners_insurance": round(monthly_insurance, 2),
"pmi": round(monthly_pmi, 2),
"hoa": monthly_hoa,
"total": round(monthly_payment, 2)
},
"total_payment_over_term": round(total_payment, 2),
"total_interest_paid": round(total_interest, 2)
}
return {
"message": f"Monthly payment for ${home_price:,} home with {down_payment_percent:.1f}% down: ${result['monthly_payment']['total']:,.2f}",
"calculation": result
}
@server.tool()
def check_health() -> Dict[str, Any]:
"""
Verify the Zillow API connection and get server status.
Returns:
Health check status
"""
logger.info("Performing health check")
if not ZILLOW_API_KEY:
return {
"error": "Zillow API key not configured",
"status": "unhealthy"
}
try:
# Make a simple API request to check connectivity
result = zillow_api_request("health")
status = {
"status": "healthy",
"api_connected": True,
"api_version": result.get("version", "unknown"),
"server_version": "1.0.0",
"rate_limit": {
"remaining": result.get("rateLimit", {}).get("remaining", "unknown"),
"limit": result.get("rateLimit", {}).get("limit", "unknown"),
"reset": result.get("rateLimit", {}).get("reset", "unknown")
}
}
return {
"message": "Zillow API is responsive and server is healthy",
"status": status
}
except ZillowAPIError as e:
return {
"error": f"API connection error: {str(e)}",
"status": {
"status": "unhealthy",
"api_connected": False,
"error": str(e)
}
}
except Exception as e:
return {
"error": f"Server error: {str(e)}",
"status": {
"status": "unhealthy",
"error": str(e)
}
}
@server.tool()
def get_server_tools() -> Dict[str, Any]:
"""
Get a list of all available tools on this server.
Returns:
List of available tools and their descriptions
"""
tools = [
{
"name": "search_properties",
"description": "Search for properties based on various criteria",
"parameters": [
{"name": "location", "type": "string", "required": True, "description": "Location to search (city, neighborhood, zip code, etc.)"},
{"name": "type", "type": "string", "required": False, "default": "forSale", "description": "Type of listing (forSale, forRent, sold, recentlySold)"},
{"name": "min_price", "type": "integer", "required": False, "description": "Minimum price"},
{"name": "max_price", "type": "integer", "required": False, "description": "Maximum price"},
{"name": "beds_min", "type": "integer", "required": False, "description": "Minimum number of bedrooms"},
{"name": "beds_max", "type": "integer", "required": False, "description": "Maximum number of bedrooms"},
{"name": "baths_min", "type": "number", "required": False, "description": "Minimum number of bathrooms"},
{"name": "baths_max", "type": "number", "required": False, "description": "Maximum number of bathrooms"},
{"name": "home_types", "type": "array", "required": False, "description": "List of home types (e.g., house, apartment, condo)"}
]
},
{
"name": "get_property_details",
"description": "Get detailed information about a specific property",
"parameters": [
{"name": "property_id", "type": "string", "required": False, "description": "Zillow property ID"},
{"name": "address", "type": "string", "required": False, "description": "Property address (used if property_id is not provided)"}
]
},
{
"name": "get_zestimate",
"description": "Get Zillow's estimated value for a property",
"parameters": [
{"name": "property_id", "type": "string", "required": False, "description": "Zillow property ID"},
{"name": "address", "type": "string", "required": False, "description": "Property address (used if property_id is not provided)"}
]
},
{
"name": "get_market_trends",
"description": "Get real estate market trends for a specific location",
"parameters": [
{"name": "location", "type": "string", "required": True, "description": "Location to get trends for (city, neighborhood, zip code, etc.)"},
{"name": "metrics", "type": "array", "required": False, "default": ["median_list_price", "median_sale_price", "median_days_on_market"], "description": "List of metrics to retrieve"},
{"name": "time_period", "type": "string", "required": False, "default": "1year", "description": "Time period for trends (1month, 3months, 6months, 1year, 5years, 10years)"}
]
},
{
"name": "calculate_mortgage",
"description": "Calculate mortgage payments and related costs",
"parameters": [
{"name": "home_price", "type": "integer", "required": True, "description": "Home purchase price"},
{"name": "down_payment", "type": "integer", "required": False, "description": "Down payment amount in dollars"},
{"name": "down_payment_percent", "type": "number", "required": False, "description": "Down payment as a percentage of home price"},
{"name": "loan_term", "type": "integer", "required": False, "default": 30, "description": "Loan term in years"},
{"name": "interest_rate", "type": "number", "required": False, "default": 6.5, "description": "Annual interest rate (percentage)"},
{"name": "annual_property_tax", "type": "integer", "required": False, "description": "Annual property tax amount"},
{"name": "annual_homeowners_insurance", "type": "integer", "required": False, "description": "Annual homeowners insurance cost"},
{"name": "monthly_hoa", "type": "integer", "required": False, "default": 0, "description": "Monthly HOA fees"},
{"name": "include_pmi", "type": "boolean", "required": False, "default": True, "description": "Whether to include PMI for down payments < 20%"}
]
},
{
"name": "check_health",
"description": "Verify the Zillow API connection and get server status",
"parameters": []
},
{
"name": "get_server_tools",
"description": "Get a list of all available tools on this server",
"parameters": []
}
]
return {
"message": "Available tools on the Zillow MCP server",
"tools": tools
}
# Define resources
@server.resource("zillow://property/{property_id}")
def property_resource(property_id: str) -> str:
"""
Get property information as a formatted text resource.
Args:
property_id: Zillow property ID
Returns:
Formatted property information
"""
try:
params = {"propertyId": property_id}
result = zillow_api_request("properties/detail", params)
property_data = result.get("property", {})
# Format the property details as text
address = property_data.get("address", {}).get("full", "Unknown address")
price = property_data.get("price", "Unknown price")
beds = property_data.get("beds", "Unknown")
baths = property_data.get("baths", "Unknown")
sqft = property_data.get("livingArea", "Unknown")
home_type = property_data.get("homeType", "Unknown")
year_built = property_data.get("yearBuilt", "Unknown")
description = property_data.get("description", "No description available")
content = f"""
# {address}
**Price:** ${price:,}
**Beds:** {beds}
**Baths:** {baths}
**Square Feet:** {sqft}
**Type:** {home_type}
**Year Built:** {year_built}
## Description
{description}
## Features
"""
features = property_data.get("features", [])
for feature_category, feature_items in features.items():
content += f"\n### {feature_category}\n"
for item in feature_items:
content += f"- {item}\n"
content += "\n## Schools\n"
schools = property_data.get("schools", [])
for school in schools:
content += f"- {school.get('name')} ({school.get('level')}): Rating {school.get('rating')}/10\n"
content += "\n## Price History\n"
price_history = property_data.get("priceHistory", [])
for event in price_history:
content += f"- {event.get('date')}: ${event.get('price'):,} ({event.get('event')})\n"
content += f"\nZestimate: ${property_data.get('zestimate', 'Unknown')}"
return content
except Exception as e:
return f"Error retrieving property information: {str(e)}"
@server.resource("zillow://market-trends/{location}")
def market_trends_resource(location: str) -> str:
"""
Get market trends information as a formatted text resource.
Args:
location: Location to get trends for
Returns:
Formatted market trends information
"""
try:
params = {
"location": location,
"metrics": "median_list_price,median_sale_price,median_days_on_market,inventory",
"period": "1year"
}
result = zillow_api_request("market/trends", params)
trends_data = result.get("trends", {})
content = f"""
# Real Estate Market Trends: {location}
## Current Market Snapshot
"""
for metric, values in trends_data.items():
current = values.get("current")
change_ytd = values.get("changeYTD")
change_mom = values.get("changeMoM")
metric_name = metric.replace("_", " ").title()
if "price" in metric:
content += f"**{metric_name}:** ${current:,}"
elif "days" in metric:
content += f"**{metric_name}:** {current} days"
else:
content += f"**{metric_name}:** {current}"
if change_ytd is not None:
content += f" (Year-to-date change: {change_ytd:+.1f}%)"
if change_mom is not None:
content += f" (Month-over-month change: {change_mom:+.1f}%)"
content += "\n\n"
content += """
## Market Analysis
This data represents the current real estate market conditions in the specified location.
The median list price indicates the middle point of all listing prices, while the median sale price
shows the middle point of actual sold properties. Days on market reflects how quickly properties are selling.
## Interpretation Guide
- **Rising prices with low days on market** indicates a seller's market with high demand
- **Falling prices with high days on market** indicates a buyer's market with lower demand
- **Stable prices with moderate days on market** indicates a balanced market
Data provided by Zillow. Last updated: """ + result.get("lastUpdated", "Unknown")
return content
except Exception as e:
return f"Error retrieving market trends: {str(e)}"
def main():
"""Main entry point for the server."""
parser = argparse.ArgumentParser(description="Zillow MCP Server")
parser.add_argument("--http", action="store_true", help="Run as HTTP server")
parser.add_argument("--port", type=int, default=8000, help="Port for HTTP server")
parser.add_argument("--debug", action="store_true", help="Enable debug logging")
args = parser.parse_args()
if args.debug:
logger.setLevel(logging.DEBUG)
logger.debug("Debug logging enabled")
if not ZILLOW_API_KEY:
logger.warning("ZILLOW_API_KEY not found. Please set it in .env file or environment variables.")
if args.http:
import uvicorn
from mcp.server.http import create_http_app
app = create_http_app(server)
logger.info(f"Starting HTTP server on port {args.port}")
uvicorn.run(app, host="0.0.0.0", port=args.port)
else:
logger.info("Starting MCP server in stdio mode")
server.run()
if __name__ == "__main__":
main()