diff --git a/README.md b/README.md index f5ad864..401e2bf 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Israel Real Estate MCP -A Python-based Mission Control Program (MCP) for interacting with the Israeli government's public real estate data API (Govmap). This tool allows real estate agents and professionals to query recent property deals based on addresses, search for properties, and retrieve detailed market information. +A Python-based Mission Control Program (MCP) for interacting with the Israeli government's public real estate data API (Govmap). This project provides both a Python library and an MCP server that allows real estate agents, professionals, and AI agents to query recent property deals, analyze market trends, and retrieve detailed real estate information. ## Description @@ -62,7 +62,196 @@ This project provides a comprehensive Python interface to the Israeli government ## Usage -### Basic Usage +### MCP Server (Recommended for AI Agents) + +The project includes an MCP (Model Context Protocol) server that allows AI agents to access Israeli real estate data. The server provides multiple deployment options: + +#### 1. Interactive Demo Server + +For testing and demonstration purposes: + +```bash +python simple_mcp_server.py +``` + +This runs an interactive demo where you can test the tools directly in the terminal. + +#### 2. Full MCP Server + +For production use with MCP clients: + +```bash +python run_mcp_server.py +``` + +This starts the full MCP server that can be connected to by MCP-compatible clients. + +#### 3. Direct Server Module + +You can also run the server directly: + +```bash +python -m nadlan_mcp.mcp_server +``` + +#### MCP Client Configuration + +To connect to the server from MCP clients, use the following configuration: + +**For Claude Desktop or other MCP clients:** + +Add to your MCP client configuration: + +```json +{ + "servers": { + "nadlan-mcp": { + "command": "python", + "args": ["/path/to/nadlan-mcp/run_mcp_server.py"], + "env": {} + } + } +} +``` + +**For development with stdio transport:** + +```python +import asyncio +from mcp.client.stdio import stdio_client + +async def main(): + async with stdio_client(["python", "run_mcp_server.py"]) as client: + # List available tools + result = await client.list_tools() + print("Available tools:", result.tools) + + # Call a tool + result = await client.call_tool("find_recent_deals_for_address", { + "address": "סוקולוב 38 חולון", + "years_back": 2 + }) + print("Results:", result) + +asyncio.run(main()) +``` + +#### Available MCP Tools + +The server provides these tools for AI agents: + +##### 🏠 `find_recent_deals_for_address` +**Main comprehensive analysis tool** +- **Description**: Find all relevant real estate deals for a given address +- **Parameters**: + - `address` (required): The address to search for (Hebrew or English) + - `years_back` (optional): Number of years to look back (default: 2) +- **Returns**: List of deals with detailed information + +**Example usage:** +```json +{ + "address": "בן יהודה 1 תל אביב", + "years_back": 3 +} +``` + +##### 📊 `analyze_market_trends` +**Market trend analysis with price insights** +- **Description**: Analyze price trends and market data for an address +- **Parameters**: + - `address` (required): The address to analyze + - `years_back` (optional): Number of years to analyze (default: 2) +- **Returns**: Market analysis with trends, average prices, and insights + +**Example usage:** +```json +{ + "address": "דיזנגוף 50 תל אביב", + "years_back": 5 +} +``` + +##### 🏘️ `compare_neighborhoods` +**Compare multiple areas** +- **Description**: Compare real estate data across multiple addresses or neighborhoods +- **Parameters**: + - `addresses` (required): List of addresses to compare + - `years_back` (optional): Number of years to analyze (default: 2) +- **Returns**: Comparative analysis with rankings and insights + +**Example usage:** +```json +{ + "addresses": ["סוקולוב 38 חולון", "בן יהודה 1 תל אביב", "דיזנגוף 50 תל אביב"], + "years_back": 2 +} +``` + +##### 🔍 `autocomplete_address` +**Address search and validation** +- **Description**: Search for and validate Israeli addresses +- **Parameters**: + - `search_text` (required): The address text to search for +- **Returns**: List of matching addresses with coordinates + +**Example usage:** +```json +{ + "search_text": "רוטשילד תל אביב" +} +``` + +#### Server Features + +- **Async Operations**: All tools run asynchronously for better performance +- **Error Handling**: Comprehensive error handling with meaningful error messages +- **Logging**: Detailed logging for debugging and monitoring +- **Data Validation**: Input validation and sanitization +- **Hebrew Support**: Full support for Hebrew addresses and text + +#### Testing the Server + +You can test the server using the interactive demo: + +```bash +python simple_mcp_server.py +``` + +This will start an interactive session where you can: +1. List available tools +2. Call tools with parameters +3. See formatted results +4. Test error handling + +#### Server Logs + +The server provides detailed logging. To see debug information: + +```bash +# Set logging level before running +export PYTHONPATH=/path/to/nadlan-mcp +python -c "import logging; logging.basicConfig(level=logging.DEBUG)" run_mcp_server.py +``` + +#### Troubleshooting + +**Connection Issues:** +- Ensure all dependencies are installed: `pip install -r requirements.txt` +- Check that Python path is correct in client configuration +- Verify the server starts without errors + +**Tool Execution Issues:** +- Check network connectivity for API calls +- Verify address format (Hebrew addresses work best) +- Review server logs for detailed error information + +**Performance:** +- Use appropriate `years_back` values (2-5 years recommended) +- Large radius searches may take longer +- Consider caching results for repeated queries + +### Python Library Usage ```python from nadlan_mcp import GovmapClient @@ -269,7 +458,9 @@ logging.basicConfig(level=logging.DEBUG) ## Dependencies - **requests**: HTTP library for API calls -- **python-dotenv**: Environment variable management (for future configuration) +- **python-dotenv**: Environment variable management +- **mcp**: Model Context Protocol SDK for AI agent integration +- **pytest**: Testing framework ## License diff --git a/nadlan_mcp/mcp_server.py b/nadlan_mcp/mcp_server.py new file mode 100644 index 0000000..33d5bd5 --- /dev/null +++ b/nadlan_mcp/mcp_server.py @@ -0,0 +1,557 @@ +""" +Israel Real Estate MCP Server + +An MCP server for accessing Israeli government real estate data through the Govmap API. +Provides tools for real estate agents and AI assistants to query property deals and market data. +""" + +import logging +from typing import Any, Dict, List, Optional +from datetime import datetime, timedelta + +from mcp.server.models import InitializationOptions +from mcp.server import NotificationOptions, Server +from mcp.types import ( + CallToolRequest, + CallToolResult, + ListToolsRequest, + ListToolsResult, + Tool, + TextContent, +) +import mcp.types as types +import mcp.server.stdio + +from .main import GovmapClient + +# Configure logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Initialize the server +server = Server("nadlan-mcp") + +# Global client instance +govmap_client = GovmapClient() + + +@server.list_tools() +async def handle_list_tools() -> ListToolsResult: + """List available MCP tools for Israeli real estate data.""" + return ListToolsResult( + tools=[ + Tool( + name="autocomplete_address", + description="Search for Israeli addresses using autocomplete. Returns coordinates and address details.", + inputSchema={ + "type": "object", + "properties": { + "search_text": { + "type": "string", + "description": "Address to search for (Hebrew or English, e.g., 'בן יהודה 1 תל אביב' or 'Ben Yehuda 1 Tel Aviv')" + } + }, + "required": ["search_text"] + } + ), + Tool( + name="get_deals_by_radius", + description="Find real estate deals within a specified radius of coordinates.", + inputSchema={ + "type": "object", + "properties": { + "longitude": { + "type": "number", + "description": "Longitude coordinate" + }, + "latitude": { + "type": "number", + "description": "Latitude coordinate" + }, + "radius": { + "type": "integer", + "description": "Search radius in meters (default: 50)", + "default": 50 + } + }, + "required": ["longitude", "latitude"] + } + ), + Tool( + name="get_street_deals", + description="Get detailed real estate deals for a specific street/polygon.", + inputSchema={ + "type": "object", + "properties": { + "polygon_id": { + "type": "string", + "description": "Polygon ID for the street/area" + }, + "limit": { + "type": "integer", + "description": "Maximum number of deals to return (default: 10)", + "default": 10 + }, + "start_date": { + "type": "string", + "description": "Start date in YYYY-MM format", + "pattern": "^\\d{4}-\\d{2}$" + }, + "end_date": { + "type": "string", + "description": "End date in YYYY-MM format", + "pattern": "^\\d{4}-\\d{2}$" + } + }, + "required": ["polygon_id"] + } + ), + Tool( + name="get_neighborhood_deals", + description="Get real estate deals within the same neighborhood as a given polygon.", + inputSchema={ + "type": "object", + "properties": { + "polygon_id": { + "type": "string", + "description": "Polygon ID for the area" + }, + "limit": { + "type": "integer", + "description": "Maximum number of deals to return (default: 10)", + "default": 10 + }, + "start_date": { + "type": "string", + "description": "Start date in YYYY-MM format", + "pattern": "^\\d{4}-\\d{2}$" + }, + "end_date": { + "type": "string", + "description": "End date in YYYY-MM format", + "pattern": "^\\d{4}-\\d{2}$" + } + }, + "required": ["polygon_id"] + } + ), + Tool( + name="find_recent_deals_for_address", + description="🏠 MAIN TOOL: Find all recent real estate deals for a given address. This is the primary function that combines all other tools to provide comprehensive market analysis.", + inputSchema={ + "type": "object", + "properties": { + "address": { + "type": "string", + "description": "Full address to search for (Hebrew or English, e.g., 'דיזנגוף 1 תל אביב')" + }, + "years_back": { + "type": "integer", + "description": "How many years back to search for deals (default: 2)", + "default": 2, + "minimum": 1, + "maximum": 10 + } + }, + "required": ["address"] + } + ), + Tool( + name="analyze_market_trends", + description="📊 Analyze market trends for a specific address including price trends, average prices, and market insights.", + inputSchema={ + "type": "object", + "properties": { + "address": { + "type": "string", + "description": "Address to analyze market trends for" + }, + "years_back": { + "type": "integer", + "description": "How many years of data to analyze (default: 3)", + "default": 3, + "minimum": 1, + "maximum": 10 + } + }, + "required": ["address"] + } + ), + Tool( + name="compare_neighborhoods", + description="🏘️ Compare real estate market data between multiple addresses/neighborhoods.", + inputSchema={ + "type": "object", + "properties": { + "addresses": { + "type": "array", + "items": {"type": "string"}, + "description": "List of addresses to compare", + "minItems": 2, + "maxItems": 5 + }, + "years_back": { + "type": "integer", + "description": "Years of data to compare (default: 2)", + "default": 2 + } + }, + "required": ["addresses"] + } + ) + ] + ) + + +@server.call_tool() +async def handle_call_tool(request: CallToolRequest) -> CallToolResult: + """Handle MCP tool calls for Israeli real estate data.""" + + try: + tool_name = request.params.name + arguments = request.params.arguments or {} + + if tool_name == "autocomplete_address": + search_text = arguments.get("search_text") + if not search_text: + raise ValueError("search_text is required") + + result = govmap_client.autocomplete_address(search_text) + + # Format the response for better readability + formatted_results = [] + for item in result.get("results", []): + formatted_results.append({ + "text": item.get("text"), + "type": item.get("type"), + "coordinates": item.get("shape"), + "score": item.get("score") + }) + + return CallToolResult( + content=[ + TextContent( + type="text", + text=f"Found {result.get('resultsCount', 0)} address matches:\n" + + "\n".join([f"• {r['text']} (type: {r['type']}, score: {r['score']})" + for r in formatted_results[:5]]) + ) + ], + isError=False + ) + + elif tool_name == "get_deals_by_radius": + longitude = arguments.get("longitude") + latitude = arguments.get("latitude") + radius = arguments.get("radius", 50) + + if longitude is None or latitude is None: + raise ValueError("longitude and latitude are required") + + result = govmap_client.get_deals_by_radius((longitude, latitude), radius) + + return CallToolResult( + content=[ + TextContent( + type="text", + text=f"Found {len(result)} deals within {radius}m radius:\n" + + "\n".join([f"• Settlement: {deal.get('settlementNameHeb', 'N/A')}, Polygon: {deal.get('polygon_id', 'N/A')}" + for deal in result[:10]]) + ) + ], + isError=False + ) + + elif tool_name == "get_street_deals": + polygon_id = arguments.get("polygon_id") + limit = arguments.get("limit", 10) + start_date = arguments.get("start_date") + end_date = arguments.get("end_date") + + if not polygon_id: + raise ValueError("polygon_id is required") + + result = govmap_client.get_street_deals(polygon_id, limit, start_date, end_date) + + deals_summary = [] + for deal in result[:5]: + 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' + deals_summary.append(f"• {date}: {price:,} NIS, {area} m²" if isinstance(price, (int, float)) else f"• {date}: {price}, {area} m²") + + return CallToolResult( + content=[ + TextContent( + type="text", + text=f"Found {len(result)} street deals for polygon {polygon_id}:\n" + "\n".join(deals_summary) + ) + ], + isError=False + ) + + elif tool_name == "get_neighborhood_deals": + polygon_id = arguments.get("polygon_id") + limit = arguments.get("limit", 10) + start_date = arguments.get("start_date") + end_date = arguments.get("end_date") + + if not polygon_id: + raise ValueError("polygon_id is required") + + result = govmap_client.get_neighborhood_deals(polygon_id, limit, start_date, end_date) + + deals_summary = [] + for deal in result[:5]: + 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' + neighborhood = deal.get('neighborhood', 'N/A') + deals_summary.append(f"• {date}: {price:,} NIS, {area} m² in {neighborhood}" if isinstance(price, (int, float)) else f"• {date}: {price}, {area} m² in {neighborhood}") + + return CallToolResult( + content=[ + TextContent( + type="text", + text=f"Found {len(result)} neighborhood deals for polygon {polygon_id}:\n" + "\n".join(deals_summary) + ) + ], + isError=False + ) + + elif tool_name == "find_recent_deals_for_address": + address = arguments.get("address") + years_back = arguments.get("years_back", 2) + + if not address: + raise ValueError("address is required") + + result = govmap_client.find_recent_deals_for_address(address, years_back) + + # Create comprehensive summary + if result: + # Calculate statistics + amounts = [deal.get('dealAmount') for deal in result if isinstance(deal.get('dealAmount'), (int, float))] + areas = [deal.get('assetArea') for deal in result if isinstance(deal.get('assetArea'), (int, float))] + + summary = [f"🏠 REAL ESTATE ANALYSIS FOR: {address}"] + summary.append(f"📊 Total deals found: {len(result)}") + + if amounts: + avg_price = sum(amounts) / len(amounts) + summary.append(f"💰 Average price: {avg_price:,.0f} NIS") + summary.append(f"📈 Price range: {min(amounts):,} - {max(amounts):,} NIS") + + if areas: + avg_area = sum(areas) / len(areas) + summary.append(f"📏 Average area: {avg_area:.0f} m²") + + summary.append(f"\n🏡 Recent deals (last {years_back} years):") + + # Show first 10 deals + for i, deal in enumerate(result[: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)): + summary.append(f"{i}. {date} | {price:,} NIS | {area} m² | {prop_type} | {neighborhood}") + else: + summary.append(f"{i}. {date} | {price} | {area} m² | {prop_type} | {neighborhood}") + + if len(result) > 10: + summary.append(f"\n... and {len(result) - 10} more deals") + + else: + summary = [f"No recent deals found for address: {address}"] + + return CallToolResult( + content=[ + TextContent( + type="text", + text="\n".join(summary) + ) + ], + isError=False + ) + + elif tool_name == "analyze_market_trends": + address = arguments.get("address") + years_back = arguments.get("years_back", 3) + + if not address: + raise ValueError("address is required") + + # Get deals data + deals = govmap_client.find_recent_deals_for_address(address, years_back) + + if not deals: + return CallToolResult( + content=[TextContent(type="text", text=f"No market data found for {address}")], + isError=False + ) + + # 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 CallToolResult( + content=[ + TextContent( + type="text", + text="\n".join(analysis) + ) + ], + isError=False + ) + + elif tool_name == "compare_neighborhoods": + addresses = arguments.get("addresses", []) + years_back = arguments.get("years_back", 2) + + if len(addresses) < 2: + raise ValueError("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 = govmap_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 CallToolResult( + content=[ + TextContent( + type="text", + text="\n".join(comparison) + ) + ], + isError=False + ) + + else: + raise ValueError(f"Unknown tool: {tool_name}") + + except Exception as e: + logger.error(f"Error in tool {request.params.name}: {str(e)}") + return CallToolResult( + content=[ + TextContent( + type="text", + text=f"Error: {str(e)}" + ) + ], + isError=True + ) + + +async def main(): + """Run the MCP server.""" + + async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): + await server.run( + read_stream, + write_stream, + InitializationOptions( + server_name="israel-real-estate-mcp", + server_version="1.0.0", + capabilities=server.get_capabilities( + notification_options=NotificationOptions(), + experimental_capabilities={}, + ), + ), + ) + + +if __name__ == "__main__": + import asyncio + asyncio.run(main()) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 3c76890..99c9d2e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ requests>=2.31.0 python-dotenv>=1.0.0 pytest>=7.0.0 +mcp>=1.0.0 types-requests>=2.31.0 diff --git a/run_mcp_server.py b/run_mcp_server.py new file mode 100644 index 0000000..6fc8d45 --- /dev/null +++ b/run_mcp_server.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +""" +Run the Israel Real Estate MCP Server + +Usage: + python run_mcp_server.py +""" + +import asyncio +import sys +import os + +# Add the current directory to the Python path +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from nadlan_mcp.mcp_server import main + +if __name__ == "__main__": + print("🏠 Starting Israel Real Estate MCP Server...") + print("🔗 Connect your AI agent to this server to access Israeli real estate data") + print("📊 Available tools: address search, market analysis, neighborhood comparison") + print("=" * 70) + + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("\n👋 Server stopped by user") + except Exception as e: + print(f"❌ Server error: {e}") + sys.exit(1) \ No newline at end of file diff --git a/simple_mcp_server.py b/simple_mcp_server.py new file mode 100644 index 0000000..7db552e --- /dev/null +++ b/simple_mcp_server.py @@ -0,0 +1,311 @@ +#!/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.main 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