diff --git a/README.md b/README.md index 45e4881..85926dc 100644 --- a/README.md +++ b/README.md @@ -86,15 +86,15 @@ python simple_mcp_server.py This runs an interactive demo where you can test the tools directly in the terminal. -#### 3. Full MCP Server (Legacy) +#### 3. Alternative: Interactive Demo -For production use with MCP clients: +For testing and demonstrations: ```bash -python run_mcp_server.py +python simple_mcp_server.py ``` -This starts the full MCP server that can be connected to by MCP-compatible clients. +This runs an interactive demo where you can test the tools directly in the terminal. #### 4. Direct Server Module @@ -124,19 +124,7 @@ Add to your MCP client configuration: } ``` -**Alternative (Legacy MCP Server):** -```json -{ - "servers": { - "nadlan-mcp": { - "command": "python", - "args": ["/path/to/nadlan-mcp/run_mcp_server.py"], - "env": {} - } - } -} -``` **For development with stdio transport (FastMCP):** @@ -162,14 +150,16 @@ asyncio.run(main()) #### Available MCP Tools -**FastMCP Server provides these 5 tools:** +**FastMCP Server provides these 7 tools:** - 🏠 `find_recent_deals_for_address` - Main comprehensive analysis tool - 📊 `get_deals_by_radius` - Find deals within a radius of coordinates - 🏘️ `get_street_deals` - Get deals for a specific street polygon +- 🏘️ `get_neighborhood_deals` - Get deals for a specific neighborhood polygon - 🔍 `autocomplete_address` - Address search and validation -- 📈 `compare_addresses` - Compare multiple addresses +- 📈 `analyze_market_trends` - Analyze market trends and price patterns +- 📊 `compare_addresses` - Compare multiple addresses -**Legacy MCP Server provides these 4 tools:** +**All Tools Details:** ##### 🏠 `find_recent_deals_for_address` **Main comprehensive analysis tool** diff --git a/nadlan_mcp/fastmcp_server.py b/nadlan_mcp/fastmcp_server.py deleted file mode 100644 index 933d481..0000000 --- a/nadlan_mcp/fastmcp_server.py +++ /dev/null @@ -1,417 +0,0 @@ -#!/usr/bin/env python3 -""" -FastMCP Server for Israeli Real Estate Data (Nadlan) - -This server provides access to Israeli government real estate data through the Govmap API -using the FastMCP library for better compatibility and reliability. -""" - -import logging -from typing import List, Dict, Any, Optional -from mcp.server.fastmcp import FastMCP -from .main import GovmapClient # type: ignore - -# Configure logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -# Initialize FastMCP server -mcp = FastMCP("nadlan-mcp") - -# Initialize the Govmap client -client = GovmapClient() - -@mcp.tool() -async def autocomplete_address(search_text: str) -> str: - """Search and autocomplete Israeli addresses. - - Args: - search_text: The partial address to search for (in Hebrew or English) - - Returns: - JSON string containing matching addresses with their coordinates - """ - try: - response = client.autocomplete_address(search_text) - - if not response or 'results' not in response: - return f"No addresses found for '{search_text}'" - - # Format results for better readability - formatted_results = [] - for result in response['results']: - formatted_results.append({ - "address": result.get("addressLabel", ""), - "settlement": result.get("settlementNameHeb", ""), - "coordinates": result.get("coordinates", {}), - "polygon_id": result.get("polygon_id") - }) - - import json - return json.dumps(formatted_results, ensure_ascii=False, indent=2) - - except Exception as e: - logger.error(f"Error in autocomplete_address: {e}") - return f"Error searching for address: {str(e)}" - -@mcp.tool() -async def get_deals_by_radius( - latitude: float, - longitude: float, - radius_meters: int = 500, - limit: int = 100 -) -> str: - """Get real estate deals within a radius of coordinates. - - Args: - latitude: Latitude coordinate - longitude: Longitude coordinate - radius_meters: Search radius in meters (default: 500) - limit: Maximum number of deals to return (default: 100) - - Returns: - JSON string containing recent real estate deals in the area - """ - try: - deals = client.get_deals_by_radius((longitude, latitude), radius_meters) - - if not deals: - return f"No deals found within {radius_meters}m of coordinates ({latitude}, {longitude})" - - # Format deals for better readability - formatted_deals = [] - for deal in deals: - formatted_deals.append({ - "address": deal.get("addressLabel", ""), - "settlement": deal.get("settlementNameHeb", ""), - "deal_amount": deal.get("dealAmount"), - "deal_date": deal.get("dealDate", ""), - "asset_area": deal.get("assetArea"), - "asset_type": deal.get("assetTypeHeb", ""), - "coordinates": deal.get("coordinates", {}) - }) - - import json - return json.dumps({ - "total_deals": len(formatted_deals), - "search_radius_meters": radius_meters, - "center_coordinates": {"latitude": latitude, "longitude": longitude}, - "deals": formatted_deals - }, ensure_ascii=False, indent=2) - - except Exception as e: - logger.error(f"Error in get_deals_by_radius: {e}") - return f"Error fetching deals by radius: {str(e)}" - -@mcp.tool() -async def get_street_deals(street_name: str, settlement_name: str, limit: int = 100) -> str: - """Get real estate deals for a specific street. - - Args: - street_name: Name of the street (in Hebrew) - settlement_name: Name of the city/settlement (in Hebrew) - limit: Maximum number of deals to return (default: 100) - - Returns: - JSON string containing recent real estate deals on the specified street - """ - try: - deals = await client.get_street_deals(street_name, settlement_name, limit) - - if not deals: - return f"No deals found for {street_name}, {settlement_name}" - - # Format deals for better readability - formatted_deals = [] - for deal in deals: - formatted_deals.append({ - "address": deal.get("addressLabel", ""), - "settlement": deal.get("settlementNameHeb", ""), - "deal_amount": deal.get("dealAmount"), - "deal_date": deal.get("dealDate", ""), - "asset_area": deal.get("assetArea"), - "asset_type": deal.get("assetTypeHeb", ""), - "coordinates": deal.get("coordinates", {}) - }) - - import json - return json.dumps({ - "total_deals": len(formatted_deals), - "street": street_name, - "settlement": settlement_name, - "deals": formatted_deals - }, ensure_ascii=False, indent=2) - - except Exception as e: - logger.error(f"Error in get_street_deals: {e}") - return f"Error fetching street deals: {str(e)}" - -@mcp.tool() -async def get_neighborhood_deals(polygon_id: str, limit: int = 100) -> str: - """Get real estate deals for a specific neighborhood polygon. - - Args: - polygon_id: The polygon ID of the neighborhood - limit: Maximum number of deals to return (default: 100) - - Returns: - JSON string containing recent real estate deals in the specified neighborhood - """ - try: - deals = await client.get_neighborhood_deals(polygon_id, limit) - - if not deals: - return f"No deals found for polygon ID {polygon_id}" - - # Format deals for better readability - formatted_deals = [] - for deal in deals: - formatted_deals.append({ - "address": deal.get("addressLabel", ""), - "settlement": deal.get("settlementNameHeb", ""), - "deal_amount": deal.get("dealAmount"), - "deal_date": deal.get("dealDate", ""), - "asset_area": deal.get("assetArea"), - "asset_type": deal.get("assetTypeHeb", ""), - "coordinates": deal.get("coordinates", {}) - }) - - import json - return json.dumps({ - "total_deals": len(formatted_deals), - "polygon_id": polygon_id, - "deals": formatted_deals - }, ensure_ascii=False, indent=2) - - except Exception as e: - logger.error(f"Error in get_neighborhood_deals: {e}") - return f"Error fetching neighborhood deals: {str(e)}" - -@mcp.tool() -async def find_recent_deals_for_address( - address: str, - radius_meters: int = 500, - limit: int = 100 -) -> str: - """Comprehensive analysis of recent real estate deals for an address. - - This is the main tool for getting detailed market analysis around a specific address. - It finds the coordinates for the address and then searches for deals in the area. - - Args: - address: The address to analyze (in Hebrew or English) - radius_meters: Search radius in meters (default: 500) - limit: Maximum number of deals to return (default: 100) - - Returns: - JSON string containing comprehensive real estate analysis including: - - Address details and coordinates - - Recent deals in the area - - Market statistics and trends - """ - try: - deals = await client.find_recent_deals_for_address(address, radius_meters, limit) - - if not deals: - return f"No deals found for address '{address}'" - - # Calculate market statistics - prices = [deal.get("dealAmount", 0) for deal in deals if deal.get("dealAmount")] - areas = [deal.get("assetArea", 0) for deal in deals if deal.get("assetArea")] - - stats = {} - if prices: - stats["price_stats"] = { - "average_price": sum(prices) / len(prices), - "min_price": min(prices), - "max_price": max(prices), - "total_deals": len(prices) - } - - if areas: - stats["area_stats"] = { - "average_area": sum(areas) / len(areas), - "min_area": min(areas), - "max_area": max(areas) - } - - # Format deals for better readability - formatted_deals = [] - for deal in deals: - formatted_deals.append({ - "address": deal.get("addressLabel", ""), - "settlement": deal.get("settlementNameHeb", ""), - "deal_amount": deal.get("dealAmount"), - "deal_date": deal.get("dealDate", ""), - "asset_area": deal.get("assetArea"), - "asset_type": deal.get("assetTypeHeb", ""), - "coordinates": deal.get("coordinates", {}) - }) - - import json - return json.dumps({ - "search_address": address, - "search_radius_meters": radius_meters, - "total_deals": len(formatted_deals), - "market_statistics": stats, - "deals": formatted_deals - }, ensure_ascii=False, indent=2) - - except Exception as e: - logger.error(f"Error in find_recent_deals_for_address: {e}") - return f"Error analyzing address: {str(e)}" - -@mcp.tool() -async def analyze_market_trends( - address: str, - radius_meters: int = 1000, - limit: int = 200 -) -> str: - """Analyze market trends and price patterns for an area. - - Args: - address: The address to analyze trends around - radius_meters: Search radius in meters (default: 1000) - limit: Maximum number of deals to analyze (default: 200) - - Returns: - JSON string containing market trend analysis including: - - Price trends over time - - Average prices by property type - - Market activity levels - """ - try: - # Get deals for the address - deals = await client.find_recent_deals_for_address(address, radius_meters, limit) - - if not deals: - return f"No deals found for market analysis near '{address}'" - - # Analyze trends by year - from collections import defaultdict - import json - - trends_by_year = defaultdict(list) - trends_by_type = defaultdict(list) - - for deal in deals: - deal_date = deal.get("dealDate", "") - deal_amount = deal.get("dealAmount", 0) - asset_type = deal.get("assetTypeHeb", "Unknown") - - if deal_date and deal_amount: - year = deal_date.split('-')[0] if '-' in deal_date else deal_date[:4] - trends_by_year[year].append(deal_amount) - trends_by_type[asset_type].append(deal_amount) - - # Calculate yearly trends - yearly_trends = {} - for year, prices in trends_by_year.items(): - yearly_trends[year] = { - "average_price": sum(prices) / len(prices), - "min_price": min(prices), - "max_price": max(prices), - "deal_count": len(prices) - } - - # Calculate type trends - type_trends = {} - for asset_type, prices in trends_by_type.items(): - type_trends[asset_type] = { - "average_price": sum(prices) / len(prices), - "min_price": min(prices), - "max_price": max(prices), - "deal_count": len(prices) - } - - return json.dumps({ - "analysis_address": address, - "analysis_radius_meters": radius_meters, - "total_deals_analyzed": len(deals), - "yearly_trends": yearly_trends, - "property_type_trends": type_trends - }, ensure_ascii=False, indent=2) - - except Exception as e: - logger.error(f"Error in analyze_market_trends: {e}") - return f"Error analyzing market trends: {str(e)}" - -@mcp.tool() -async def compare_neighborhoods(addresses: List[str], radius_meters: int = 500) -> str: - """Compare real estate markets between multiple neighborhoods. - - Args: - addresses: List of addresses to compare (in Hebrew or English) - radius_meters: Search radius for each address (default: 500) - - Returns: - JSON string containing comparative analysis of multiple neighborhoods - """ - try: - import json - - comparisons = [] - - for address in addresses: - try: - deals = await client.find_recent_deals_for_address(address, radius_meters, 100) - - if deals: - prices = [] - areas = [] - for deal in deals: - if isinstance(deal, dict): - if deal.get("dealAmount"): - prices.append(deal.get("dealAmount", 0)) - if deal.get("assetArea"): - areas.append(deal.get("assetArea", 0)) - - comparison = { - "address": address, - "total_deals": len(deals), - "price_stats": { - "average_price": sum(prices) / len(prices) if prices else 0, - "min_price": min(prices) if prices else 0, - "max_price": max(prices) if prices else 0 - }, - "area_stats": { - "average_area": sum(areas) / len(areas) if areas else 0, - "min_area": min(areas) if areas else 0, - "max_area": max(areas) if areas else 0 - } - } - else: - comparison = { - "address": address, - "total_deals": 0, - "price_stats": {}, - "area_stats": {} - } - - comparisons.append(comparison) - - except Exception as e: - logger.error(f"Error comparing {address}: {e}") - comparisons.append({ - "address": address, - "error": str(e) - }) - - # Rank neighborhoods by average price - valid_comparisons = [c for c in comparisons if c.get("price_stats", {}).get("average_price", 0) > 0] - valid_comparisons.sort(key=lambda x: x["price_stats"]["average_price"], reverse=True) - - return json.dumps({ - "comparison_radius_meters": radius_meters, - "neighborhoods_compared": len(addresses), - "ranking_by_average_price": valid_comparisons, - "all_results": comparisons - }, ensure_ascii=False, indent=2) - - except Exception as e: - logger.error(f"Error in compare_neighborhoods: {e}") - return f"Error comparing neighborhoods: {str(e)}" - -# Run the server -if __name__ == "__main__": - mcp.run() \ No newline at end of file diff --git a/nadlan_mcp/mcp_server.py b/nadlan_mcp/mcp_server.py deleted file mode 100644 index c0e6538..0000000 --- a/nadlan_mcp/mcp_server.py +++ /dev/null @@ -1,557 +0,0 @@ -""" -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="nadlan-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/nadlan_mcp/simple_fastmcp_server.py b/nadlan_mcp/simple_fastmcp_server.py index 0276b37..a9759fe 100644 --- a/nadlan_mcp/simple_fastmcp_server.py +++ b/nadlan_mcp/simple_fastmcp_server.py @@ -8,7 +8,7 @@ using the FastMCP library with simplified, working functions. import json import logging -from typing import List +from typing import List, Dict from mcp.server.fastmcp import FastMCP from .main import GovmapClient @@ -160,6 +160,128 @@ def find_recent_deals_for_address(address: str, years_back: int = 2) -> str: logger.error(f"Error in find_recent_deals_for_address: {e}") return f"Error analyzing address: {str(e)}" +@mcp.tool() +def get_neighborhood_deals(polygon_id: str, limit: int = 100) -> str: + """Get real estate deals for a specific neighborhood polygon. + + Args: + polygon_id: The polygon ID of the neighborhood + limit: Maximum number of deals to return (default: 100) + + Returns: + JSON string containing recent real estate deals in the specified neighborhood + """ + try: + deals = client.get_neighborhood_deals(polygon_id, limit) + + if not deals: + return f"No deals found for polygon ID {polygon_id}" + + return json.dumps({ + "total_deals": len(deals), + "polygon_id": polygon_id, + "deals": deals + }, ensure_ascii=False, indent=2) + + except Exception as e: + logger.error(f"Error in get_neighborhood_deals: {e}") + return f"Error fetching neighborhood deals: {str(e)}" + +@mcp.tool() +def analyze_market_trends(address: str, years_back: int = 3) -> str: + """Analyze market trends and price patterns for an area. + + Args: + address: The address to analyze trends around + years_back: How many years of data to analyze (default: 3) + + Returns: + JSON string containing market trend analysis including: + - Price trends over time + - Average prices by property type + - Market activity levels + - Price per square meter trends + """ + try: + # Get deals for the address + deals = client.find_recent_deals_for_address(address, years_back) + + if not deals: + return f"No deals found for market analysis near '{address}'" + + # Analyze trends by year + from collections import defaultdict + + yearly_data = defaultdict(list) + property_types: Dict[str, int] = defaultdict(int) + 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('assetTypeHeb', deal.get('propertyTypeDescription', 'Unknown')) + neighborhood = deal.get('settlementNameHeb', deal.get('neighborhood')) + + if neighborhood: + neighborhoods.add(neighborhood) + + if isinstance(price, (int, float)) and isinstance(area, (int, float)) and area > 0: + yearly_data[year].append({ + 'price': price, + 'area': area, + 'price_per_sqm': price / area + }) + property_types[prop_type] += 1 + + # Calculate yearly trends + yearly_trends = {} + for year, year_deals in yearly_data.items(): + if year_deals: + yearly_trends[year] = { + "average_price": sum(d['price'] for d in year_deals) / len(year_deals), + "min_price": min(d['price'] for d in year_deals), + "max_price": max(d['price'] for d in year_deals), + "average_area": sum(d['area'] for d in year_deals) / len(year_deals), + "average_price_per_sqm": sum(d['price_per_sqm'] for d in year_deals) / len(year_deals), + "deal_count": len(year_deals) + } + + # Calculate price trend direction + price_trend_analysis = {} + years_sorted = sorted(yearly_trends.keys()) + if len(years_sorted) >= 2: + first_year_avg = yearly_trends[years_sorted[0]]['average_price_per_sqm'] + last_year_avg = yearly_trends[years_sorted[-1]]['average_price_per_sqm'] + + trend_percentage = ((last_year_avg - first_year_avg) / first_year_avg) * 100 + trend_direction = "rising" if trend_percentage > 5 else "declining" if trend_percentage < -5 else "stable" + + price_trend_analysis = { + "trend_direction": trend_direction, + "trend_percentage": round(trend_percentage, 1), + "first_year": years_sorted[0], + "last_year": years_sorted[-1], + "first_year_avg_price_per_sqm": round(first_year_avg, 0), + "last_year_avg_price_per_sqm": round(last_year_avg, 0) + } + + return json.dumps({ + "analysis_address": address, + "analysis_period_years": years_back, + "total_deals_analyzed": len(deals), + "neighborhoods": list(neighborhoods), + "property_types": dict(property_types), + "yearly_trends": yearly_trends, + "price_trend_analysis": price_trend_analysis + }, ensure_ascii=False, indent=2) + + except Exception as e: + logger.error(f"Error in analyze_market_trends: {e}") + return f"Error analyzing market trends: {str(e)}" + @mcp.tool() def compare_addresses(addresses: List[str]) -> str: """Compare real estate markets between multiple addresses. diff --git a/run_mcp_server.py b/run_mcp_server.py deleted file mode 100644 index 6fc8d45..0000000 --- a/run_mcp_server.py +++ /dev/null @@ -1,30 +0,0 @@ -#!/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