Nitzan Pomerantz
311 lines
14 KiB
Python
311 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple Israel Real Estate MCP Server
|
|
|
|
A simplified MCP server that provides access to Israeli real estate data.
|
|
Focuses on the main use cases for AI agents.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from typing import Any, Dict, List
|
|
|
|
# Simple MCP server implementation
|
|
class SimpleMCPServer:
|
|
def __init__(self):
|
|
from nadlan_mcp.govmap import GovmapClient
|
|
self.client = GovmapClient()
|
|
self.tools = {
|
|
"find_recent_deals_for_address": {
|
|
"description": "🏠 Find all recent real estate deals for a given address. Main function for comprehensive market analysis.",
|
|
"parameters": {
|
|
"address": "Full address to search for (Hebrew or English)",
|
|
"years_back": "How many years back to search (default: 2)"
|
|
}
|
|
},
|
|
"analyze_market_trends": {
|
|
"description": "📊 Analyze market trends for a specific address with price analysis and insights",
|
|
"parameters": {
|
|
"address": "Address to analyze",
|
|
"years_back": "Years of data to analyze (default: 3)"
|
|
}
|
|
},
|
|
"compare_neighborhoods": {
|
|
"description": "🏘️ Compare real estate market data between multiple addresses",
|
|
"parameters": {
|
|
"addresses": "List of addresses to compare (2-5 addresses)",
|
|
"years_back": "Years of data to compare (default: 2)"
|
|
}
|
|
},
|
|
"autocomplete_address": {
|
|
"description": "🔍 Search for Israeli addresses using autocomplete",
|
|
"parameters": {
|
|
"search_text": "Address to search for"
|
|
}
|
|
}
|
|
}
|
|
|
|
def list_tools(self) -> str:
|
|
"""List available tools."""
|
|
tools_list = ["📋 Available Real Estate Tools:"]
|
|
for name, info in self.tools.items():
|
|
tools_list.append(f"\n🔧 {name}")
|
|
tools_list.append(f" {info['description']}")
|
|
tools_list.append(" Parameters:")
|
|
for param, desc in info['parameters'].items():
|
|
tools_list.append(f" • {param}: {desc}")
|
|
|
|
return "\n".join(tools_list)
|
|
|
|
def call_tool(self, tool_name: str, **kwargs) -> str:
|
|
"""Call a specific tool."""
|
|
try:
|
|
if tool_name == "find_recent_deals_for_address":
|
|
return self._find_recent_deals(**kwargs)
|
|
elif tool_name == "analyze_market_trends":
|
|
return self._analyze_trends(**kwargs)
|
|
elif tool_name == "compare_neighborhoods":
|
|
return self._compare_neighborhoods(**kwargs)
|
|
elif tool_name == "autocomplete_address":
|
|
return self._autocomplete_address(**kwargs)
|
|
else:
|
|
return f"❌ Unknown tool: {tool_name}"
|
|
except Exception as e:
|
|
return f"❌ Error in {tool_name}: {str(e)}"
|
|
|
|
def _find_recent_deals(self, address: str, years_back: int = 2) -> str:
|
|
"""Find recent deals for an address."""
|
|
deals = self.client.find_recent_deals_for_address(address, years_back)
|
|
|
|
if not deals:
|
|
return f"No recent deals found for: {address}"
|
|
|
|
# Calculate statistics
|
|
amounts = [deal.get('dealAmount') for deal in deals if isinstance(deal.get('dealAmount'), (int, float))]
|
|
areas = [deal.get('assetArea') for deal in deals if isinstance(deal.get('assetArea'), (int, float))]
|
|
|
|
result = [f"🏠 REAL ESTATE ANALYSIS FOR: {address}"]
|
|
result.append(f"📊 Total deals found: {len(deals)}")
|
|
|
|
if amounts:
|
|
avg_price = sum(amounts) / len(amounts)
|
|
result.append(f"💰 Average price: {avg_price:,.0f} NIS")
|
|
result.append(f"📈 Price range: {min(amounts):,} - {max(amounts):,} NIS")
|
|
|
|
if areas:
|
|
avg_area = sum(areas) / len(areas)
|
|
result.append(f"📏 Average area: {avg_area:.0f} m²")
|
|
|
|
result.append(f"\n🏡 Recent deals (last {years_back} years):")
|
|
|
|
# Show first 10 deals
|
|
for i, deal in enumerate(deals[:10], 1):
|
|
price = deal.get('dealAmount', 'N/A')
|
|
area = deal.get('assetArea', 'N/A')
|
|
date = deal.get('dealDate', 'N/A')[:10] if deal.get('dealDate') else 'N/A'
|
|
prop_type = deal.get('propertyTypeDescription', 'N/A')
|
|
neighborhood = deal.get('neighborhood', 'N/A')
|
|
|
|
if isinstance(price, (int, float)):
|
|
result.append(f"{i}. {date} | {price:,} NIS | {area} m² | {prop_type} | {neighborhood}")
|
|
else:
|
|
result.append(f"{i}. {date} | {price} | {area} m² | {prop_type} | {neighborhood}")
|
|
|
|
if len(deals) > 10:
|
|
result.append(f"\n... and {len(deals) - 10} more deals")
|
|
|
|
return "\n".join(result)
|
|
|
|
def _analyze_trends(self, address: str, years_back: int = 3) -> str:
|
|
"""Analyze market trends."""
|
|
deals = self.client.find_recent_deals_for_address(address, years_back)
|
|
|
|
if not deals:
|
|
return f"No market data found for {address}"
|
|
|
|
# Analyze trends by year
|
|
yearly_data = {}
|
|
property_types = {}
|
|
neighborhoods = set()
|
|
|
|
for deal in deals:
|
|
date_str = deal.get('dealDate', '')
|
|
if date_str:
|
|
year = date_str[:4]
|
|
price = deal.get('dealAmount')
|
|
area = deal.get('assetArea')
|
|
prop_type = deal.get('propertyTypeDescription', 'Unknown')
|
|
neighborhood = deal.get('neighborhood')
|
|
|
|
if neighborhood:
|
|
neighborhoods.add(neighborhood)
|
|
|
|
if isinstance(price, (int, float)) and isinstance(area, (int, float)) and area > 0:
|
|
if year not in yearly_data:
|
|
yearly_data[year] = []
|
|
yearly_data[year].append({
|
|
'price': price,
|
|
'area': area,
|
|
'price_per_sqm': price / area
|
|
})
|
|
|
|
property_types[prop_type] = property_types.get(prop_type, 0) + 1
|
|
|
|
# Generate analysis
|
|
analysis = [f"📊 MARKET TRENDS ANALYSIS: {address}"]
|
|
analysis.append(f"📅 Analysis period: Last {years_back} years")
|
|
analysis.append(f"🏘️ Neighborhoods: {', '.join(neighborhoods) if neighborhoods else 'N/A'}")
|
|
analysis.append(f"🏠 Property types: {', '.join([f'{k} ({v})' for k, v in property_types.items()])}")
|
|
|
|
if yearly_data:
|
|
analysis.append(f"\n📈 YEARLY TRENDS:")
|
|
|
|
for year in sorted(yearly_data.keys(), reverse=True):
|
|
year_deals = yearly_data[year]
|
|
avg_price = sum(d['price'] for d in year_deals) / len(year_deals)
|
|
avg_area = sum(d['area'] for d in year_deals) / len(year_deals)
|
|
avg_price_per_sqm = sum(d['price_per_sqm'] for d in year_deals) / len(year_deals)
|
|
|
|
analysis.append(f" {year}: {len(year_deals)} deals | Avg: {avg_price:,.0f} NIS | {avg_area:.0f} m² | {avg_price_per_sqm:,.0f} NIS/m²")
|
|
|
|
# Price trend
|
|
years_sorted = sorted(yearly_data.keys())
|
|
if len(years_sorted) >= 2:
|
|
first_year_avg = sum(d['price_per_sqm'] for d in yearly_data[years_sorted[0]]) / len(yearly_data[years_sorted[0]])
|
|
last_year_avg = sum(d['price_per_sqm'] for d in yearly_data[years_sorted[-1]]) / len(yearly_data[years_sorted[-1]])
|
|
|
|
trend = ((last_year_avg - first_year_avg) / first_year_avg) * 100
|
|
trend_direction = "📈 Rising" if trend > 0 else "📉 Declining" if trend < 0 else "➡️ Stable"
|
|
analysis.append(f"\n🎯 Price Trend: {trend_direction} ({trend:+.1f}% over period)")
|
|
|
|
return "\n".join(analysis)
|
|
|
|
def _compare_neighborhoods(self, addresses: List[str], years_back: int = 2) -> str:
|
|
"""Compare neighborhoods."""
|
|
if len(addresses) < 2:
|
|
return "❌ At least 2 addresses are required for comparison"
|
|
|
|
comparison = [f"🏘️ NEIGHBORHOOD COMPARISON"]
|
|
comparison.append(f"📅 Comparing last {years_back} years of data\n")
|
|
|
|
address_data = {}
|
|
|
|
for address in addresses:
|
|
deals = self.client.find_recent_deals_for_address(address, years_back)
|
|
|
|
if deals:
|
|
amounts = [deal.get('dealAmount') for deal in deals if isinstance(deal.get('dealAmount'), (int, float))]
|
|
areas = [deal.get('assetArea') for deal in deals if isinstance(deal.get('assetArea'), (int, float))]
|
|
neighborhoods = {deal.get('neighborhood') for deal in deals if deal.get('neighborhood')}
|
|
|
|
if amounts and areas:
|
|
price_per_sqm = [amounts[i] / areas[i] for i in range(min(len(amounts), len(areas))) if areas[i] > 0]
|
|
|
|
address_data[address] = {
|
|
'deals_count': len(deals),
|
|
'avg_price': sum(amounts) / len(amounts),
|
|
'avg_area': sum(areas) / len(areas),
|
|
'avg_price_per_sqm': sum(price_per_sqm) / len(price_per_sqm) if price_per_sqm else 0,
|
|
'neighborhoods': neighborhoods
|
|
}
|
|
|
|
# Generate comparison
|
|
for address, data in address_data.items():
|
|
comparison.append(f"📍 {address}:")
|
|
comparison.append(f" • {data['deals_count']} deals found")
|
|
comparison.append(f" • Avg price: {data['avg_price']:,.0f} NIS")
|
|
comparison.append(f" • Avg area: {data['avg_area']:.0f} m²")
|
|
comparison.append(f" • Price per m²: {data['avg_price_per_sqm']:,.0f} NIS")
|
|
comparison.append(f" • Neighborhoods: {', '.join(data['neighborhoods'])}")
|
|
comparison.append("")
|
|
|
|
# Ranking
|
|
if address_data:
|
|
comparison.append("🏆 RANKINGS:")
|
|
by_price_per_sqm = sorted(address_data.items(), key=lambda x: x[1]['avg_price_per_sqm'], reverse=True)
|
|
|
|
comparison.append("💰 Most expensive (NIS/m²):")
|
|
for i, (addr, data) in enumerate(by_price_per_sqm, 1):
|
|
comparison.append(f" {i}. {addr}: {data['avg_price_per_sqm']:,.0f} NIS/m²")
|
|
|
|
return "\n".join(comparison)
|
|
|
|
def _autocomplete_address(self, search_text: str) -> str:
|
|
"""Autocomplete address search."""
|
|
result = self.client.autocomplete_address(search_text)
|
|
|
|
formatted_results = []
|
|
for item in result.get("results", []):
|
|
formatted_results.append(f"• {item.get('text')} (type: {item.get('type')}, score: {item.get('score')})")
|
|
|
|
return f"Found {result.get('resultsCount', 0)} address matches:\n" + "\n".join(formatted_results[:5])
|
|
|
|
|
|
def main():
|
|
"""Interactive demo of the MCP server functionality."""
|
|
print("🏠 Israel Real Estate MCP Server - Demo Mode")
|
|
print("=" * 50)
|
|
|
|
server = SimpleMCPServer()
|
|
|
|
while True:
|
|
print("\n" + "=" * 50)
|
|
print("🔧 Available commands:")
|
|
print("1. list - List all available tools")
|
|
print("2. find - Find recent deals for address")
|
|
print("3. trends - Analyze market trends")
|
|
print("4. compare - Compare neighborhoods")
|
|
print("5. search - Search for addresses")
|
|
print("6. quit - Exit")
|
|
|
|
choice = input("\nSelect a command (1-6): ").strip()
|
|
|
|
if choice == "1":
|
|
print(server.list_tools())
|
|
|
|
elif choice == "2":
|
|
address = input("Enter address: ").strip()
|
|
years = input("Years back (default 2): ").strip()
|
|
years_back = int(years) if years.isdigit() else 2
|
|
|
|
print("\n🔍 Searching for deals...")
|
|
result = server.call_tool("find_recent_deals_for_address", address=address, years_back=years_back)
|
|
print(result)
|
|
|
|
elif choice == "3":
|
|
address = input("Enter address: ").strip()
|
|
years = input("Years back (default 3): ").strip()
|
|
years_back = int(years) if years.isdigit() else 3
|
|
|
|
print("\n📊 Analyzing trends...")
|
|
result = server.call_tool("analyze_market_trends", address=address, years_back=years_back)
|
|
print(result)
|
|
|
|
elif choice == "4":
|
|
addresses_input = input("Enter addresses (comma separated): ").strip()
|
|
addresses = [addr.strip() for addr in addresses_input.split(",")]
|
|
years = input("Years back (default 2): ").strip()
|
|
years_back = int(years) if years.isdigit() else 2
|
|
|
|
print("\n🏘️ Comparing neighborhoods...")
|
|
result = server.call_tool("compare_neighborhoods", addresses=addresses, years_back=years_back)
|
|
print(result)
|
|
|
|
elif choice == "5":
|
|
search_text = input("Enter search text: ").strip()
|
|
|
|
print("\n🔍 Searching addresses...")
|
|
result = server.call_tool("autocomplete_address", search_text=search_text)
|
|
print(result)
|
|
|
|
elif choice == "6":
|
|
print("👋 Goodbye!")
|
|
break
|
|
|
|
else:
|
|
print("❌ Invalid choice. Please select 1-6.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |