From b844760147790991aa6da60889d118f7cbc24091 Mon Sep 17 00:00:00 2001 From: Nitzan Pomerantz <9297302+nitzpo@users.noreply.github.com> Date: Sun, 19 Oct 2025 00:58:27 +0300 Subject: [PATCH] Remove old file --- mcp_server_concept.py | 311 ------------------------------------------ 1 file changed, 311 deletions(-) delete mode 100644 mcp_server_concept.py diff --git a/mcp_server_concept.py b/mcp_server_concept.py deleted file mode 100644 index 1ec9f3a..0000000 --- a/mcp_server_concept.py +++ /dev/null @@ -1,311 +0,0 @@ -#!/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() \ No newline at end of file