mcp-servers/nadlan-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mcp clean up

Browse Source
This commit is contained in:
Nitzan Pomerantz
2025-07-13 00:56:47 +03:00
parent d8c18cb2f7
commit a1fa75a878
5 changed files with 132 additions and 1024 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+9 -19
View File
@@ -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**
nadlan_mcp/fastmcp_server.py
-417
View File
@@ -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()
nadlan_mcp/mcp_server.py
-557
View File
@@ -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())
nadlan_mcp/simple_fastmcp_server.py
+123 -1
View File
@@ -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.
run_mcp_server.py
-30
View File
@@ -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)