chaim
4bc054f315
The PDF host (free-justice.openapi.gov.il) requires the same x-client-id header as the search API, so a normal browser click on the URL fails. This tool carries the auth header automatically and saves the PDF to a configurable local directory. Safety properties: - SSRF guard: only download from free-justice.openapi.gov.il and pub-justice.openapi.gov.il. - Path-traversal guard: filename is reduced to its basename; arbitrary paths are stripped. - Content-type guard: rejects 200-OK responses whose body is not a real PDF (gateway sometimes returns JSON-error 200s). - Atomic write via .tmp + rename so partial downloads never replace the cached copy. - Caches by destination path; re-downloads are no-ops unless overwrite=True. DECISIVE_APPRAISER_DOWNLOAD_DIR env var configures the output dir (default: ./downloads/decisive_appraisals). 6 new unit tests cover the happy path, caching, and all four guards. End-to-end live test confirmed a 1.4MB real PDF lands on disk with valid `%PDF-1.7` header. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1487 lines
59 KiB
Python
1487 lines
59 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Simple FastMCP Server for Israeli Real Estate Data
|
|
|
|
This server provides access to Israeli government real estate data through the Govmap API
|
|
using the FastMCP library with simplified, working functions.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
from starlette.responses import JSONResponse
|
|
|
|
from nadlan_mcp.config import get_config
|
|
from nadlan_mcp.govil import DecisiveAppraiserClient
|
|
from nadlan_mcp.govmap import GovmapClient
|
|
from nadlan_mcp.govmap.models import Deal
|
|
from nadlan_mcp.govmap.outlier_detection import filter_deals_for_analysis
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Initialize FastMCP server
|
|
mcp = FastMCP("nadlan-mcp")
|
|
|
|
# Initialize the Govmap client
|
|
client = GovmapClient()
|
|
|
|
# Initialize the Decisive Appraiser (gov.il / Ministry of Justice) client
|
|
decisive_appraiser_client = DecisiveAppraiserClient()
|
|
|
|
|
|
def conditional_tool(config_flag: str):
|
|
"""
|
|
Decorator to conditionally register MCP tools based on config.
|
|
|
|
If the config flag is False, the tool won't be registered with FastMCP.
|
|
"""
|
|
|
|
def decorator(func):
|
|
config = get_config()
|
|
enabled = getattr(config, config_flag, True)
|
|
|
|
if enabled:
|
|
# Register the tool with FastMCP
|
|
return mcp.tool()(func)
|
|
else:
|
|
# Don't register, just return the function unchanged
|
|
logger.info(f"Tool {func.__name__} is DISABLED (config: {config_flag})")
|
|
return func
|
|
|
|
return decorator
|
|
|
|
|
|
def log_mcp_call(func_name: str, **params):
|
|
"""
|
|
Log MCP tool function calls with their parameters for debugging and monitoring.
|
|
|
|
Args:
|
|
func_name: Name of the MCP tool function being called
|
|
**params: Keyword arguments passed to the function
|
|
"""
|
|
# Format parameters for logging (truncate long strings)
|
|
formatted_params = {}
|
|
for key, value in params.items():
|
|
if isinstance(value, str) and len(value) > 100:
|
|
formatted_params[key] = f"{value[:97]}..."
|
|
elif isinstance(value, list) and len(value) > 5:
|
|
formatted_params[key] = f"[{len(value)} items]"
|
|
else:
|
|
formatted_params[key] = value
|
|
|
|
logger.info(
|
|
f"MCP tool called: {func_name}({', '.join(f'{k}={v}' for k, v in formatted_params.items())})"
|
|
)
|
|
|
|
|
|
def strip_bloat_fields(deals: List[Deal], lang: str = "he") -> List[Dict[str, Any]]:
|
|
"""
|
|
Remove bloat fields from Deal models to reduce token usage in MCP responses.
|
|
|
|
Converts Deal models to dictionaries and removes unnecessary fields:
|
|
- shape: Large MULTIPOLYGON coordinate data
|
|
- sourceorder, objectid, priority: Internal fields
|
|
- source_polygon_id, settlementId, streetCode, dealId, polygonId: Internal IDs
|
|
- deal_type_description: Redundant (kept in search_parameters only)
|
|
|
|
Adds sequential ID for LLM reference and selects language for text fields.
|
|
|
|
Args:
|
|
deals: List of Deal model instances
|
|
lang: Language for text values ("he" for Hebrew, "en" for English)
|
|
|
|
Returns:
|
|
List of deal dictionaries with bloat removed and sequential IDs
|
|
"""
|
|
bloat_fields = {
|
|
"shape",
|
|
"sourceorder",
|
|
"objectid",
|
|
"priority",
|
|
"source_polygon_id",
|
|
"settlementId",
|
|
"streetCode",
|
|
"dealId",
|
|
"polygonId",
|
|
"deal_type_description",
|
|
}
|
|
|
|
result = []
|
|
for idx, deal in enumerate(deals, start=1):
|
|
# Convert Deal model to dict, excluding None values
|
|
deal_dict = deal.model_dump(mode="json", exclude_none=True)
|
|
|
|
# Remove bloat fields
|
|
filtered_dict = {k: v for k, v in deal_dict.items() if k not in bloat_fields}
|
|
|
|
# Add sequential ID for LLM reference
|
|
filtered_dict["id"] = idx
|
|
|
|
# Select language for text fields (Hebrew or English)
|
|
if lang == "en":
|
|
# Use English values if available
|
|
if "settlementNameEng" in filtered_dict:
|
|
filtered_dict["settlement_name"] = filtered_dict.pop("settlementNameEng")
|
|
filtered_dict.pop("settlementNameHeb", None)
|
|
if "streetNameEng" in filtered_dict:
|
|
filtered_dict["street_name"] = filtered_dict.pop("streetNameEng")
|
|
filtered_dict.pop("streetNameHeb", None)
|
|
if "assetTypeEng" in filtered_dict:
|
|
filtered_dict["asset_type"] = filtered_dict.pop("assetTypeEng")
|
|
filtered_dict.pop("assetTypeHeb", None)
|
|
else: # Default to Hebrew
|
|
# Use Hebrew values
|
|
if "settlementNameHeb" in filtered_dict:
|
|
filtered_dict["settlement_name"] = filtered_dict.pop("settlementNameHeb")
|
|
filtered_dict.pop("settlementNameEng", None)
|
|
if "streetNameHeb" in filtered_dict:
|
|
filtered_dict["street_name"] = filtered_dict.pop("streetNameHeb")
|
|
filtered_dict.pop("streetNameEng", None)
|
|
if "assetTypeHeb" in filtered_dict:
|
|
filtered_dict["asset_type"] = filtered_dict.pop("assetTypeHeb")
|
|
filtered_dict.pop("assetTypeEng", None)
|
|
|
|
result.append(filtered_dict)
|
|
|
|
return result
|
|
|
|
|
|
@conditional_tool("tool_autocomplete_address_enabled")
|
|
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
|
|
"""
|
|
log_mcp_call("autocomplete_address", search_text=search_text)
|
|
try:
|
|
response = client.autocomplete_address(search_text)
|
|
|
|
if not response.results:
|
|
return f"No addresses found for '{search_text}'"
|
|
|
|
# Format results for better readability
|
|
formatted_results = []
|
|
for result in response.results:
|
|
result_dict = {
|
|
"text": result.text,
|
|
"id": result.id,
|
|
"type": result.type,
|
|
"score": result.score,
|
|
}
|
|
|
|
# Add coordinates if available
|
|
if result.coordinates:
|
|
result_dict["coordinates"] = {
|
|
"longitude": result.coordinates.longitude,
|
|
"latitude": result.coordinates.latitude,
|
|
}
|
|
|
|
formatted_results.append(result_dict)
|
|
|
|
return json.dumps(formatted_results, ensure_ascii=False, indent=None)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in autocomplete_address: {e}", exc_info=True)
|
|
return f"Error searching for address: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_get_deals_by_radius_enabled")
|
|
def get_deals_by_radius(latitude: float, longitude: float, radius_meters: int = 500) -> str:
|
|
"""Get polygon metadata within a radius of coordinates.
|
|
|
|
**NOTE**: This returns polygon/area metadata, NOT individual deal transactions!
|
|
Use find_recent_deals_for_address() to get actual deals.
|
|
|
|
Args:
|
|
latitude: Latitude coordinate
|
|
longitude: Longitude coordinate
|
|
radius_meters: Search radius in meters (default: 500)
|
|
|
|
Returns:
|
|
JSON string containing polygon metadata (areas with deals nearby)
|
|
"""
|
|
log_mcp_call(
|
|
"get_deals_by_radius", latitude=latitude, longitude=longitude, radius_meters=radius_meters
|
|
)
|
|
try:
|
|
# Note: GovmapClient expects (longitude, latitude) tuple
|
|
# Returns polygon metadata dicts, not Deal objects
|
|
polygons = client.get_deals_by_radius((longitude, latitude), radius_meters)
|
|
|
|
if not polygons:
|
|
return f"No polygons found within {radius_meters}m of coordinates ({latitude}, {longitude})"
|
|
|
|
return json.dumps(
|
|
{
|
|
"total_polygons": len(polygons),
|
|
"search_radius_meters": radius_meters,
|
|
"center_coordinates": {"latitude": latitude, "longitude": longitude},
|
|
"polygons": polygons, # Return dicts directly, no stripping needed
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in get_deals_by_radius: {e}", exc_info=True)
|
|
return f"Error fetching polygons by radius: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_get_street_deals_enabled")
|
|
def get_street_deals(polygon_id: str, limit: int = 100, deal_type: int = 2) -> str:
|
|
"""Get real estate deals for a specific street polygon.
|
|
|
|
Args:
|
|
polygon_id: The polygon ID of the street/area
|
|
limit: Maximum number of deals to return (default: 100)
|
|
deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2)
|
|
|
|
Returns:
|
|
JSON string containing recent real estate deals for the street
|
|
"""
|
|
log_mcp_call("get_street_deals", polygon_id=polygon_id, limit=limit, deal_type=deal_type)
|
|
try:
|
|
deals = client.get_street_deals(polygon_id, limit, deal_type=deal_type)
|
|
|
|
if not deals:
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
return f"No {deal_type_desc} deals found for polygon ID {polygon_id}"
|
|
|
|
# Add deal type metadata
|
|
for deal in deals:
|
|
deal.deal_type = deal_type
|
|
deal.deal_type_description = "first_hand_new" if deal_type == 1 else "second_hand_used"
|
|
|
|
# Calculate basic statistics using computed fields from models
|
|
price_per_sqm_values = [deal.price_per_sqm for deal in deals if deal.price_per_sqm]
|
|
|
|
stats = {}
|
|
if price_per_sqm_values:
|
|
stats["price_per_sqm_stats"] = {
|
|
"average_price_per_sqm": round(
|
|
sum(price_per_sqm_values) / len(price_per_sqm_values), 0
|
|
),
|
|
"min_price_per_sqm": round(min(price_per_sqm_values), 0),
|
|
"max_price_per_sqm": round(max(price_per_sqm_values), 0),
|
|
}
|
|
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
return json.dumps(
|
|
{
|
|
"total_deals": len(deals),
|
|
"polygon_id": polygon_id,
|
|
"deal_type": deal_type,
|
|
"deal_type_description": deal_type_desc,
|
|
"market_statistics": stats,
|
|
"deals": strip_bloat_fields(deals, lang="he"),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in get_street_deals: {e}", exc_info=True)
|
|
return f"Error fetching street deals: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_find_recent_deals_for_address_enabled")
|
|
def find_recent_deals_for_address(
|
|
address: str,
|
|
years_back: int = 2,
|
|
radius_meters: int = 50,
|
|
max_deals: int = 100,
|
|
deal_type: int = 2,
|
|
lang: str = "he",
|
|
) -> str:
|
|
"""Find recent real estate deals for a specific address.
|
|
|
|
Args:
|
|
address: The address to search for (in Hebrew or English)
|
|
years_back: How many years back to search (default: 2)
|
|
radius_meters: Search radius in meters from the address (default: 50)
|
|
Deals beyond this radius are filtered out, EXCEPT same-building deals
|
|
which are always included regardless of distance.
|
|
max_deals: Maximum number of deals to return (default: 100, provides good context for LLM analysis)
|
|
deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2)
|
|
lang: Language for text values ("he" for Hebrew [default], "en" for English)
|
|
|
|
Returns:
|
|
JSON string containing recent real estate deals within radius_meters of the address
|
|
"""
|
|
log_mcp_call(
|
|
"find_recent_deals_for_address",
|
|
address=address,
|
|
years_back=years_back,
|
|
radius_meters=radius_meters,
|
|
max_deals=max_deals,
|
|
deal_type=deal_type,
|
|
)
|
|
try:
|
|
# Get coordinates for search center
|
|
autocomplete_result = client.autocomplete_address(address)
|
|
search_coords = None
|
|
if autocomplete_result.results:
|
|
coords = autocomplete_result.results[0].coordinates
|
|
if coords:
|
|
search_coords = {"longitude": coords.longitude, "latitude": coords.latitude}
|
|
|
|
deals = client.find_recent_deals_for_address(
|
|
address, years_back, radius_meters, max_deals, deal_type
|
|
)
|
|
|
|
if not deals:
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
return f"No {deal_type_desc} deals found for address '{address}'"
|
|
|
|
# Calculate comprehensive statistics using model attributes
|
|
prices = [deal.deal_amount for deal in deals if deal.deal_amount]
|
|
areas = [deal.asset_area for deal in deals if deal.asset_area]
|
|
price_per_sqm_values = [deal.price_per_sqm for deal in deals if deal.price_per_sqm]
|
|
|
|
# Separate building, street and neighborhood deals for analysis
|
|
# deal_source is added dynamically in find_recent_deals_for_address
|
|
building_deals = [
|
|
deal for deal in deals if getattr(deal, "deal_source", None) == "same_building"
|
|
]
|
|
street_deals = [deal for deal in deals if getattr(deal, "deal_source", None) == "street"]
|
|
neighborhood_deals = [
|
|
deal for deal in deals if getattr(deal, "deal_source", None) == "neighborhood"
|
|
]
|
|
|
|
stats = {
|
|
"deal_breakdown": {
|
|
"total_deals": len(deals),
|
|
"same_building_deals": len(building_deals),
|
|
"street_deals": len(street_deals),
|
|
"neighborhood_deals": len(neighborhood_deals),
|
|
"same_building_percentage": round((len(building_deals) / len(deals)) * 100, 1)
|
|
if deals
|
|
else 0,
|
|
"street_emphasis_percentage": round((len(street_deals) / len(deals)) * 100, 1)
|
|
if deals
|
|
else 0,
|
|
"neighborhood_percentage": round((len(neighborhood_deals) / len(deals)) * 100, 1)
|
|
if deals
|
|
else 0,
|
|
}
|
|
}
|
|
|
|
# Standardize field names to match other tools
|
|
if prices:
|
|
stats["price_statistics"] = {
|
|
"mean": round(sum(prices) / len(prices), 0),
|
|
"min": min(prices),
|
|
"max": max(prices),
|
|
"median": sorted(prices)[len(prices) // 2] if prices else 0,
|
|
"total": sum(prices),
|
|
}
|
|
|
|
if areas:
|
|
stats["area_statistics"] = {
|
|
"mean": round(sum(areas) / len(areas), 1),
|
|
"min": min(areas),
|
|
"max": max(areas),
|
|
"median": sorted(areas)[len(areas) // 2] if areas else 0,
|
|
}
|
|
|
|
if price_per_sqm_values:
|
|
stats["price_per_sqm_statistics"] = {
|
|
"mean": round(sum(price_per_sqm_values) / len(price_per_sqm_values), 0),
|
|
"min": round(min(price_per_sqm_values), 0),
|
|
"max": round(max(price_per_sqm_values), 0),
|
|
"median": round(sorted(price_per_sqm_values)[len(price_per_sqm_values) // 2], 0)
|
|
if price_per_sqm_values
|
|
else 0,
|
|
}
|
|
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
search_params = {
|
|
"address": address,
|
|
"years_back": years_back,
|
|
"radius_meters": radius_meters,
|
|
"max_deals": max_deals,
|
|
"deal_type": deal_type,
|
|
"deal_type_description": deal_type_desc,
|
|
}
|
|
if search_coords:
|
|
search_params["search_coordinates"] = search_coords
|
|
|
|
return json.dumps(
|
|
{
|
|
"search_parameters": search_params,
|
|
"market_statistics": stats,
|
|
"deals": strip_bloat_fields(deals, lang=lang),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in find_recent_deals_for_address: {e}", exc_info=True)
|
|
return f"Error analyzing address: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_get_neighborhood_deals_enabled")
|
|
def get_neighborhood_deals(
|
|
polygon_id: str, limit: int = 100, deal_type: int = 2, lang: str = "he"
|
|
) -> 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)
|
|
deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2)
|
|
lang: Language for text values ("he" for Hebrew [default], "en" for English)
|
|
|
|
Returns:
|
|
JSON string containing recent real estate deals in the specified neighborhood
|
|
"""
|
|
log_mcp_call("get_neighborhood_deals", polygon_id=polygon_id, limit=limit, deal_type=deal_type)
|
|
try:
|
|
deals = client.get_neighborhood_deals(polygon_id, limit, deal_type=deal_type)
|
|
|
|
if not deals:
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
return f"No {deal_type_desc} deals found for polygon ID {polygon_id}"
|
|
|
|
# Add deal type metadata
|
|
for deal in deals:
|
|
deal.deal_type = deal_type
|
|
deal.deal_type_description = "first_hand_new" if deal_type == 1 else "second_hand_used"
|
|
|
|
# Calculate basic statistics using computed fields from models
|
|
price_per_sqm_values = [deal.price_per_sqm for deal in deals if deal.price_per_sqm]
|
|
|
|
stats = {}
|
|
if price_per_sqm_values:
|
|
stats["price_per_sqm_stats"] = {
|
|
"average_price_per_sqm": round(
|
|
sum(price_per_sqm_values) / len(price_per_sqm_values), 0
|
|
),
|
|
"min_price_per_sqm": round(min(price_per_sqm_values), 0),
|
|
"max_price_per_sqm": round(max(price_per_sqm_values), 0),
|
|
}
|
|
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
return json.dumps(
|
|
{
|
|
"total_deals": len(deals),
|
|
"polygon_id": polygon_id,
|
|
"deal_type": deal_type,
|
|
"deal_type_description": deal_type_desc,
|
|
"market_statistics": stats,
|
|
"deals": strip_bloat_fields(deals, lang=lang),
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in get_neighborhood_deals: {e}", exc_info=True)
|
|
return f"Error fetching neighborhood deals: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_analyze_market_trends_enabled")
|
|
def analyze_market_trends(
|
|
address: str,
|
|
years_back: int = 3,
|
|
radius_meters: int = 100,
|
|
max_deals: int = 100,
|
|
deal_type: int = 2,
|
|
) -> str:
|
|
"""Analyze market trends and price patterns for an area with comprehensive data.
|
|
|
|
**BETA**: This tool is not well tested or reliable. Use with caution.
|
|
|
|
Args:
|
|
address: The address to analyze trends around
|
|
years_back: How many years of data to analyze (default: 3)
|
|
radius_meters: Search radius in meters from the address (default: 100, larger for trend analysis)
|
|
max_deals: Maximum number of deals to analyze (default: 100, optimized for performance and token limits)
|
|
deal_type: Deal type filter (1=first hand/new, 2=second hand/used, default: 2)
|
|
|
|
Returns:
|
|
JSON string containing comprehensive market trend analysis (summarized data, not raw deals)
|
|
"""
|
|
log_mcp_call(
|
|
"analyze_market_trends",
|
|
address=address,
|
|
years_back=years_back,
|
|
radius_meters=radius_meters,
|
|
max_deals=max_deals,
|
|
deal_type=deal_type,
|
|
)
|
|
try:
|
|
# Get coordinates for search center
|
|
autocomplete_result = client.autocomplete_address(address)
|
|
search_coords = None
|
|
if autocomplete_result.results:
|
|
coords = autocomplete_result.results[0].coordinates
|
|
if coords:
|
|
search_coords = {"longitude": coords.longitude, "latitude": coords.latitude}
|
|
|
|
# Get deals for the address with larger radius for trend analysis
|
|
deals = client.find_recent_deals_for_address(
|
|
address, years_back, radius_meters, max_deals, deal_type
|
|
)
|
|
|
|
if not deals:
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
return f"No {deal_type_desc} deals found for comprehensive market analysis near '{address}'"
|
|
|
|
# Efficient analysis with reduced complexity
|
|
from collections import defaultdict
|
|
|
|
yearly_data = defaultdict(list)
|
|
property_types: Dict[str, List[float]] = defaultdict(
|
|
list
|
|
) # Store only prices for efficiency
|
|
neighborhoods = defaultdict(list)
|
|
|
|
# Simplified processing - extract only essential data
|
|
for deal in deals:
|
|
if not deal.deal_date:
|
|
continue
|
|
|
|
# Convert date to string for parsing
|
|
from datetime import date as date_type
|
|
|
|
date_str = (
|
|
deal.deal_date.isoformat()
|
|
if isinstance(deal.deal_date, date_type)
|
|
else str(deal.deal_date)
|
|
)
|
|
year = date_str[:4]
|
|
price = deal.deal_amount
|
|
area = deal.asset_area
|
|
price_per_sqm = deal.price_per_sqm
|
|
prop_type = deal.property_type_description or "לא ידוע"
|
|
neighborhood = deal.settlement_name_heb or deal.neighborhood or "לא ידוע"
|
|
deal_source = getattr(deal, "deal_source", "unknown")
|
|
|
|
if (
|
|
isinstance(price, (int, float))
|
|
and isinstance(area, (int, float))
|
|
and area > 0
|
|
and isinstance(price_per_sqm, (int, float))
|
|
):
|
|
yearly_data[year].append(
|
|
{
|
|
"price": price,
|
|
"area": area,
|
|
"price_per_sqm": price_per_sqm,
|
|
"deal_source": deal_source,
|
|
}
|
|
)
|
|
property_types[prop_type].append(price_per_sqm)
|
|
neighborhoods[neighborhood].append(price_per_sqm)
|
|
|
|
# Calculate streamlined yearly trends
|
|
yearly_trends = {}
|
|
for year, year_deals in yearly_data.items():
|
|
if year_deals:
|
|
prices = [d["price"] for d in year_deals]
|
|
price_per_sqm_vals = [d["price_per_sqm"] for d in year_deals]
|
|
building_deals = [d for d in year_deals if d["deal_source"] == "same_building"]
|
|
street_deals = [d for d in year_deals if d["deal_source"] == "street"]
|
|
|
|
yearly_trends[year] = {
|
|
"deal_count": len(year_deals),
|
|
"same_building_deals": len(building_deals),
|
|
"street_deals": len(street_deals),
|
|
"avg_price": round(sum(prices) / len(prices), 0),
|
|
"avg_price_per_sqm": round(
|
|
sum(price_per_sqm_vals) / len(price_per_sqm_vals), 0
|
|
),
|
|
"min_price_per_sqm": round(min(price_per_sqm_vals), 0),
|
|
"max_price_per_sqm": round(max(price_per_sqm_vals), 0),
|
|
"total_volume": sum(prices),
|
|
}
|
|
|
|
# Streamlined property type analysis (top 5 only)
|
|
property_type_analysis = {}
|
|
for prop_type, prices_per_sqm in property_types.items():
|
|
if len(prices_per_sqm) >= 2: # Only include types with multiple deals
|
|
property_type_analysis[prop_type] = {
|
|
"deal_count": len(prices_per_sqm),
|
|
"avg_price_per_sqm": round(sum(prices_per_sqm) / len(prices_per_sqm), 0),
|
|
}
|
|
|
|
# Keep only top 5 property types by deal count
|
|
property_type_analysis = dict(
|
|
sorted(property_type_analysis.items(), key=lambda x: x[1]["deal_count"], reverse=True)[
|
|
:5
|
|
]
|
|
)
|
|
|
|
# Streamlined neighborhood analysis (top 5 only)
|
|
neighborhood_analysis = {}
|
|
for neighborhood, prices_per_sqm in neighborhoods.items():
|
|
if len(prices_per_sqm) >= 3: # Minimum 3 deals for statistical significance
|
|
neighborhood_analysis[neighborhood] = {
|
|
"deal_count": len(prices_per_sqm),
|
|
"avg_price_per_sqm": round(sum(prices_per_sqm) / len(prices_per_sqm), 0),
|
|
}
|
|
|
|
# Keep only top 5 neighborhoods by deal count
|
|
neighborhood_analysis = dict(
|
|
sorted(neighborhood_analysis.items(), key=lambda x: x[1]["deal_count"], reverse=True)[
|
|
:5
|
|
]
|
|
)
|
|
|
|
# Simple trend analysis
|
|
years_sorted = sorted(yearly_trends.keys())
|
|
trend_analysis = {}
|
|
if len(years_sorted) >= 2:
|
|
first_year = yearly_trends[years_sorted[0]]
|
|
last_year = yearly_trends[years_sorted[-1]]
|
|
|
|
if first_year["avg_price_per_sqm"] > 0:
|
|
price_change = (
|
|
(last_year["avg_price_per_sqm"] - first_year["avg_price_per_sqm"])
|
|
/ first_year["avg_price_per_sqm"]
|
|
) * 100
|
|
volume_change = (
|
|
(
|
|
(last_year["deal_count"] - first_year["deal_count"])
|
|
/ first_year["deal_count"]
|
|
)
|
|
* 100
|
|
if first_year["deal_count"] > 0
|
|
else 0
|
|
)
|
|
|
|
trend_analysis = {
|
|
"price_trend_percentage": round(price_change, 1),
|
|
"volume_trend_percentage": round(volume_change, 1),
|
|
"first_year_avg_price_per_sqm": first_year["avg_price_per_sqm"],
|
|
"last_year_avg_price_per_sqm": last_year["avg_price_per_sqm"],
|
|
}
|
|
|
|
deal_type_desc = "first hand (new)" if deal_type == 1 else "second hand (used)"
|
|
|
|
# Return summarized analysis (NO raw deals to save tokens)
|
|
# Normalize structure with standard market_statistics while keeping tool-specific analysis
|
|
analysis_params = {
|
|
"address": address,
|
|
"years_analyzed": years_back,
|
|
"radius_meters": radius_meters,
|
|
"deals_analyzed": len(deals),
|
|
"deal_type": deal_type,
|
|
"deal_type_description": deal_type_desc,
|
|
}
|
|
if search_coords:
|
|
analysis_params["search_coordinates"] = search_coords
|
|
|
|
return json.dumps(
|
|
{
|
|
"analysis_parameters": analysis_params,
|
|
"market_statistics": {
|
|
"deal_breakdown": {
|
|
"total_deals": len(deals),
|
|
},
|
|
},
|
|
"market_summary": {
|
|
"years_with_data": len(yearly_trends),
|
|
"unique_property_types": len(property_type_analysis),
|
|
"unique_neighborhoods": len(neighborhood_analysis),
|
|
},
|
|
"yearly_trends": yearly_trends,
|
|
"top_property_types": property_type_analysis,
|
|
"top_neighborhoods": neighborhood_analysis,
|
|
"trend_analysis": trend_analysis,
|
|
"key_insights": {
|
|
"most_active_year": max(
|
|
yearly_trends.keys(), key=lambda y: yearly_trends[y]["deal_count"]
|
|
)
|
|
if yearly_trends
|
|
else None,
|
|
"highest_avg_price_year": max(
|
|
yearly_trends.keys(), key=lambda y: yearly_trends[y]["avg_price_per_sqm"]
|
|
)
|
|
if yearly_trends
|
|
else None,
|
|
"deal_source_summary": f"Building: {len([d for d in deals if getattr(d, 'deal_source', None) == 'same_building'])}, Street: {len([d for d in deals if getattr(d, 'deal_source', None) == 'street'])}, Neighborhood: {len([d for d in deals if getattr(d, 'deal_source', None) == 'neighborhood'])}",
|
|
},
|
|
"deals": [], # Trend analysis doesn't return raw deals to save tokens
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in analyze_market_trends: {e}", exc_info=True)
|
|
return f"Error analyzing market trends: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_compare_addresses_enabled")
|
|
def compare_addresses(addresses: List[str]) -> str:
|
|
"""Compare real estate markets between multiple addresses.
|
|
|
|
**BETA**: This tool is not well tested or reliable. Use with caution.
|
|
|
|
Args:
|
|
addresses: List of addresses to compare (in Hebrew or English)
|
|
|
|
Returns:
|
|
JSON string containing comparative analysis of multiple addresses
|
|
"""
|
|
log_mcp_call("compare_addresses", addresses=addresses)
|
|
try:
|
|
comparisons = []
|
|
|
|
for address in addresses:
|
|
try:
|
|
# Get coordinates for this address
|
|
search_coords = None
|
|
try:
|
|
autocomplete_result = client.autocomplete_address(address)
|
|
if autocomplete_result.results:
|
|
coords = autocomplete_result.results[0].coordinates
|
|
if coords:
|
|
search_coords = {
|
|
"longitude": coords.longitude,
|
|
"latitude": coords.latitude,
|
|
}
|
|
except Exception:
|
|
pass # Continue without coordinates if autocomplete fails
|
|
|
|
deals = client.find_recent_deals_for_address(address, 2)
|
|
|
|
if deals:
|
|
prices = [deal.deal_amount for deal in deals if deal.deal_amount]
|
|
areas = [deal.asset_area for deal in deals if deal.asset_area]
|
|
price_per_sqm_values = [
|
|
deal.price_per_sqm for deal in deals if deal.price_per_sqm
|
|
]
|
|
building_deals = [
|
|
deal
|
|
for deal in deals
|
|
if getattr(deal, "deal_source", None) == "same_building"
|
|
]
|
|
street_deals = [
|
|
deal for deal in deals if getattr(deal, "deal_source", None) == "street"
|
|
]
|
|
neighborhood_deals = [
|
|
deal
|
|
for deal in deals
|
|
if getattr(deal, "deal_source", None) == "neighborhood"
|
|
]
|
|
|
|
comparison = {
|
|
"address": address,
|
|
"search_coordinates": search_coords,
|
|
"total_deals": len(deals),
|
|
"same_building_deals": len(building_deals),
|
|
"street_deals": len(street_deals),
|
|
"neighborhood_deals": len(neighborhood_deals),
|
|
"same_building_percentage": round(
|
|
(len(building_deals) / len(deals)) * 100, 1
|
|
)
|
|
if deals
|
|
else 0,
|
|
"street_emphasis_percentage": round(
|
|
(len(street_deals) / len(deals)) * 100, 1
|
|
)
|
|
if deals
|
|
else 0,
|
|
"price_stats": {
|
|
"average_price": round(sum(prices) / len(prices), 0) if prices else 0,
|
|
"min_price": min(prices) if prices else 0,
|
|
"max_price": max(prices) if prices else 0,
|
|
},
|
|
"area_stats": {
|
|
"average_area": round(sum(areas) / len(areas), 1) if areas else 0,
|
|
"min_area": min(areas) if areas else 0,
|
|
"max_area": max(areas) if areas else 0,
|
|
},
|
|
"price_per_sqm_stats": {
|
|
"average_price_per_sqm": round(
|
|
sum(price_per_sqm_values) / len(price_per_sqm_values), 0
|
|
)
|
|
if price_per_sqm_values
|
|
else 0,
|
|
"min_price_per_sqm": round(min(price_per_sqm_values), 0)
|
|
if price_per_sqm_values
|
|
else 0,
|
|
"max_price_per_sqm": round(max(price_per_sqm_values), 0)
|
|
if price_per_sqm_values
|
|
else 0,
|
|
},
|
|
}
|
|
else:
|
|
comparison = {
|
|
"address": address,
|
|
"search_coordinates": search_coords,
|
|
"total_deals": 0,
|
|
"same_building_deals": 0,
|
|
"street_deals": 0,
|
|
"neighborhood_deals": 0,
|
|
"same_building_percentage": 0,
|
|
"street_emphasis_percentage": 0,
|
|
"price_stats": {},
|
|
"area_stats": {},
|
|
"price_per_sqm_stats": {},
|
|
}
|
|
|
|
comparisons.append(comparison)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error comparing {address}: {e}")
|
|
comparisons.append({"address": address, "error": str(e)})
|
|
|
|
# Rank addresses by average price per sqm
|
|
valid_comparisons = []
|
|
for comparison in comparisons:
|
|
if (
|
|
isinstance(comparison, dict)
|
|
and "price_per_sqm_stats" in comparison
|
|
and isinstance(comparison["price_per_sqm_stats"], dict)
|
|
and comparison["price_per_sqm_stats"].get("average_price_per_sqm", 0) > 0
|
|
):
|
|
valid_comparisons.append(comparison)
|
|
|
|
# Sort by price per sqm
|
|
def get_price_per_sqm(comp: dict) -> float:
|
|
price_stats = comp.get("price_per_sqm_stats", {})
|
|
if isinstance(price_stats, dict):
|
|
return price_stats.get("average_price_per_sqm", 0)
|
|
return 0
|
|
|
|
valid_comparisons.sort(key=get_price_per_sqm, reverse=True)
|
|
|
|
return json.dumps(
|
|
{
|
|
"addresses_compared": len(addresses),
|
|
"ranking_by_average_price_per_sqm": valid_comparisons,
|
|
"all_results": comparisons,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in compare_addresses: {e}", exc_info=True)
|
|
return f"Error comparing addresses: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_get_valuation_comparables_enabled")
|
|
def get_valuation_comparables(
|
|
address: str,
|
|
years_back: int = 2,
|
|
property_type: Optional[str] = None,
|
|
min_rooms: Optional[float] = None,
|
|
max_rooms: Optional[float] = None,
|
|
min_price: Optional[float] = None,
|
|
max_price: Optional[float] = None,
|
|
min_area: Optional[float] = None,
|
|
max_area: Optional[float] = None,
|
|
min_floor: Optional[int] = None,
|
|
max_floor: Optional[int] = None,
|
|
radius_meters: int = 100,
|
|
max_comparables: int = 50,
|
|
iqr_multiplier: Optional[float] = None,
|
|
include_outlier_deals: bool = True,
|
|
lang: str = "he",
|
|
) -> str:
|
|
"""Get comparable properties for valuation analysis.
|
|
|
|
This tool provides detailed comparable deals filtered by your criteria.
|
|
Automatically applies IQR outlier filtering (k=1.0 default) to remove statistical outliers
|
|
and improve data quality. The response includes metadata about filtering so you can
|
|
inform users about removed outliers.
|
|
|
|
Args:
|
|
address: The address to find comparables for (in Hebrew or English)
|
|
years_back: How many years back to search (default: 2)
|
|
property_type: Filter by property type (e.g., "דירה", "בית", "פנטהאוז")
|
|
min_rooms: Minimum number of rooms
|
|
max_rooms: Maximum number of rooms
|
|
min_price: Minimum deal amount (NIS)
|
|
max_price: Maximum deal amount (NIS)
|
|
min_area: Minimum asset area (square meters)
|
|
max_area: Maximum asset area (square meters)
|
|
min_floor: Minimum floor number
|
|
max_floor: Maximum floor number
|
|
radius_meters: Search radius in meters (default: 100, larger than find_recent_deals to get more comparables)
|
|
max_comparables: Maximum number of deals to return (default: 50, optimized for MCP token limits)
|
|
iqr_multiplier: Override IQR multiplier for outlier detection (default: 1.0). Lower = more aggressive filtering
|
|
include_outlier_deals: If True (default), include the outlier deals separately in the response
|
|
This allows you to see what was filtered out and answer questions about it
|
|
lang: Language for text values ("he" for Hebrew [default], "en" for English)
|
|
|
|
Returns:
|
|
JSON string containing:
|
|
- deals: Comparable deals that passed filtering
|
|
- deal_breakdown with outlier filtering metadata:
|
|
- total_deals: Count after filtering
|
|
- total_deals_before_filtering: Count before filtering
|
|
- outliers_removed: Number of deals removed as outliers
|
|
- filtering_method: Method used (e.g., "iqr")
|
|
- iqr_multiplier: IQR multiplier used (e.g., 1.0)
|
|
- outlier_deals: List of deals that were removed as outliers (if include_outlier_deals=True)
|
|
"""
|
|
log_mcp_call(
|
|
"get_valuation_comparables",
|
|
address=address,
|
|
years_back=years_back,
|
|
property_type=property_type,
|
|
min_rooms=min_rooms,
|
|
max_rooms=max_rooms,
|
|
min_price=min_price,
|
|
max_price=max_price,
|
|
min_area=min_area,
|
|
max_area=max_area,
|
|
min_floor=min_floor,
|
|
max_floor=max_floor,
|
|
radius_meters=radius_meters,
|
|
max_comparables=max_comparables,
|
|
iqr_multiplier=iqr_multiplier,
|
|
)
|
|
try:
|
|
# Get coordinates for search center
|
|
autocomplete_result = client.autocomplete_address(address)
|
|
search_coords = None
|
|
if autocomplete_result.results:
|
|
coords = autocomplete_result.results[0].coordinates
|
|
if coords:
|
|
search_coords = {"longitude": coords.longitude, "latitude": coords.latitude}
|
|
|
|
# Get all deals for the address with higher limits for valuation
|
|
deals = client.find_recent_deals_for_address(
|
|
address, years_back, radius=radius_meters, max_deals=max_comparables
|
|
)
|
|
|
|
search_params_base = {
|
|
"address": address,
|
|
"years_back": years_back,
|
|
"radius_meters": radius_meters,
|
|
"max_comparables": max_comparables,
|
|
}
|
|
if search_coords:
|
|
search_params_base["search_coordinates"] = search_coords
|
|
|
|
if not deals:
|
|
return json.dumps(
|
|
{
|
|
"search_parameters": search_params_base,
|
|
"market_statistics": {
|
|
"deal_breakdown": {
|
|
"total_deals": 0,
|
|
},
|
|
},
|
|
"deals": [],
|
|
"message": "No deals found for this address",
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
# Apply filters
|
|
logger.info(f"Applying criteria filters to {len(deals)} deals")
|
|
filtered_deals = client.filter_deals_by_criteria(
|
|
deals,
|
|
property_type=property_type,
|
|
min_rooms=min_rooms,
|
|
max_rooms=max_rooms,
|
|
min_price=min_price,
|
|
max_price=max_price,
|
|
min_area=min_area,
|
|
max_area=max_area,
|
|
min_floor=min_floor,
|
|
max_floor=max_floor,
|
|
)
|
|
logger.info(
|
|
f"After criteria filtering: {len(filtered_deals)} deals "
|
|
f"(removed {len(deals) - len(filtered_deals)} deals)"
|
|
)
|
|
|
|
# Apply outlier filtering to remove statistical outliers
|
|
config = get_config()
|
|
outlier_report = None
|
|
if (
|
|
config.analysis_outlier_method != "none"
|
|
and len(filtered_deals) >= config.analysis_min_deals_for_outlier_detection
|
|
):
|
|
deals_before_outlier_filter = len(filtered_deals)
|
|
filtered_deals, outlier_report = filter_deals_for_analysis(
|
|
filtered_deals,
|
|
config,
|
|
metric="price_per_sqm",
|
|
iqr_multiplier=iqr_multiplier,
|
|
include_outlier_deals=include_outlier_deals,
|
|
)
|
|
effective_k = (
|
|
iqr_multiplier if iqr_multiplier is not None else config.analysis_iqr_multiplier
|
|
)
|
|
logger.info(
|
|
f"After outlier filtering ({config.analysis_outlier_method}, k={effective_k}): "
|
|
f"{len(filtered_deals)} deals (removed {deals_before_outlier_filter - len(filtered_deals)} outliers)"
|
|
)
|
|
else:
|
|
logger.info(
|
|
f"Skipping outlier filtering: only {len(filtered_deals)} deals "
|
|
f"(minimum {config.analysis_min_deals_for_outlier_detection} required)"
|
|
)
|
|
|
|
# Calculate statistics on filtered comparables
|
|
stats = client.calculate_deal_statistics(filtered_deals)
|
|
|
|
# Build deal breakdown with outlier filtering information
|
|
deal_breakdown = {
|
|
"total_deals": len(filtered_deals),
|
|
}
|
|
|
|
# Add outlier filtering metadata if filtering was applied
|
|
if outlier_report:
|
|
deal_breakdown["total_deals_before_filtering"] = outlier_report["total_deals"]
|
|
deal_breakdown["outliers_removed"] = outlier_report["outliers_removed"]
|
|
deal_breakdown["filtering_method"] = outlier_report["method_used"]
|
|
if outlier_report["method_used"] == "iqr":
|
|
deal_breakdown["iqr_multiplier"] = outlier_report["parameters"]["iqr_multiplier"]
|
|
|
|
# Build response with filtered deals
|
|
search_params_base["filters_applied"] = {
|
|
"property_type": property_type,
|
|
"rooms": f"{min_rooms}-{max_rooms}" if min_rooms or max_rooms else None,
|
|
"price": f"{min_price}-{max_price}" if min_price or max_price else None,
|
|
"area": f"{min_area}-{max_area}" if min_area or max_area else None,
|
|
"floor": f"{min_floor}-{max_floor}" if min_floor or max_floor else None,
|
|
}
|
|
|
|
response_data = {
|
|
"search_parameters": search_params_base,
|
|
"market_statistics": {
|
|
"deal_breakdown": deal_breakdown,
|
|
"price_statistics": stats.price_statistics,
|
|
"area_statistics": stats.area_statistics,
|
|
"price_per_sqm_statistics": stats.price_per_sqm_statistics,
|
|
"property_type_distribution": stats.property_type_distribution,
|
|
"date_range": stats.date_range,
|
|
},
|
|
"deals": strip_bloat_fields(filtered_deals, lang=lang),
|
|
}
|
|
|
|
# Add outlier deals if present and requested
|
|
if include_outlier_deals and outlier_report and "outlier_deals" in outlier_report:
|
|
response_data["outlier_deals"] = strip_bloat_fields(
|
|
outlier_report["outlier_deals"], lang=lang
|
|
)
|
|
|
|
return json.dumps(response_data, ensure_ascii=False, indent=None)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in get_valuation_comparables: {e}", exc_info=True)
|
|
return f"Error getting valuation comparables: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_get_deal_statistics_enabled")
|
|
def get_deal_statistics(
|
|
address: str,
|
|
years_back: int = 2,
|
|
property_type: Optional[str] = None,
|
|
min_rooms: Optional[float] = None,
|
|
max_rooms: Optional[float] = None,
|
|
iqr_multiplier: Optional[float] = None,
|
|
) -> str:
|
|
"""Calculate statistical aggregations on deal data for an address.
|
|
|
|
This tool provides quick statistical summaries WITHOUT returning any deals.
|
|
Useful when LLM needs calculations on large datasets without full deal details.
|
|
|
|
Args:
|
|
address: The address to analyze (in Hebrew or English)
|
|
years_back: How many years back to analyze (default: 2)
|
|
property_type: Filter by property type (e.g., "דירה", "בית")
|
|
min_rooms: Minimum number of rooms
|
|
max_rooms: Maximum number of rooms
|
|
iqr_multiplier: Override IQR multiplier for outlier detection (default: 1.0). Lower = more aggressive filtering
|
|
|
|
Returns:
|
|
JSON string containing market_statistics only (no deals returned)
|
|
"""
|
|
log_mcp_call(
|
|
"get_deal_statistics",
|
|
address=address,
|
|
years_back=years_back,
|
|
property_type=property_type,
|
|
min_rooms=min_rooms,
|
|
max_rooms=max_rooms,
|
|
iqr_multiplier=iqr_multiplier,
|
|
)
|
|
try:
|
|
# Get coordinates for search center
|
|
autocomplete_result = client.autocomplete_address(address)
|
|
search_coords = None
|
|
if autocomplete_result.results:
|
|
coords = autocomplete_result.results[0].coordinates
|
|
if coords:
|
|
search_coords = {"longitude": coords.longitude, "latitude": coords.latitude}
|
|
|
|
# Get all deals for the address
|
|
deals = client.find_recent_deals_for_address(address, years_back)
|
|
|
|
search_params_base = {
|
|
"address": address,
|
|
"years_back": years_back,
|
|
}
|
|
if search_coords:
|
|
search_params_base["search_coordinates"] = search_coords
|
|
|
|
if not deals:
|
|
return json.dumps(
|
|
{
|
|
"search_parameters": search_params_base,
|
|
"market_statistics": {
|
|
"deal_breakdown": {"total_deals": 0},
|
|
"message": "No deals found for this address",
|
|
},
|
|
"deals": [],
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
# Apply filters if provided
|
|
if property_type or min_rooms or max_rooms:
|
|
deals = client.filter_deals_by_criteria(
|
|
deals, property_type=property_type, min_rooms=min_rooms, max_rooms=max_rooms
|
|
)
|
|
|
|
# Calculate statistics (don't need outlier deals for statistics-only tool)
|
|
stats = client.calculate_deal_statistics(
|
|
deals, iqr_multiplier=iqr_multiplier, include_outlier_deals=False
|
|
)
|
|
|
|
# Build response - statistics only, NO deals
|
|
search_params_base["filters_applied"] = {
|
|
"property_type": property_type,
|
|
"rooms": f"{min_rooms}-{max_rooms}" if min_rooms or max_rooms else None,
|
|
}
|
|
|
|
response_data = {
|
|
"search_parameters": search_params_base,
|
|
"market_statistics": {
|
|
"deal_breakdown": {
|
|
"total_deals": stats.total_deals,
|
|
},
|
|
"price_statistics": stats.price_statistics,
|
|
"area_statistics": stats.area_statistics,
|
|
"price_per_sqm_statistics": stats.price_per_sqm_statistics,
|
|
"property_type_distribution": stats.property_type_distribution,
|
|
"date_range": stats.date_range,
|
|
},
|
|
}
|
|
|
|
return json.dumps(response_data, ensure_ascii=False, indent=None)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in get_deal_statistics: {e}", exc_info=True)
|
|
return f"Error calculating deal statistics: {str(e)}"
|
|
|
|
|
|
def _safe_calculate_metric(metric_func, deals):
|
|
"""
|
|
Safely execute a metric calculation function.
|
|
|
|
Helper function to reduce code duplication in try-except blocks
|
|
for market metric calculations.
|
|
|
|
Args:
|
|
metric_func: Function to call with deals as argument
|
|
deals: List of Deal model instances to analyze
|
|
|
|
Returns:
|
|
Result dictionary from metric_func (serialized from Pydantic model),
|
|
or error dictionary if any exception raised
|
|
"""
|
|
try:
|
|
result = metric_func(deals)
|
|
# Serialize Pydantic model to dict
|
|
if hasattr(result, "model_dump"):
|
|
return result.model_dump(exclude_none=True)
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"Error calculating metric {metric_func.__name__}: {e}")
|
|
return {"error": str(e)}
|
|
|
|
|
|
@conditional_tool("tool_get_market_activity_metrics_enabled")
|
|
def get_market_activity_metrics(address: str, years_back: int = 2, radius_meters: int = 100) -> str:
|
|
"""Get comprehensive market activity and investment potential analysis.
|
|
|
|
This tool provides detailed market liquidity, activity scores, and investment
|
|
potential metrics. It combines activity scoring, liquidity analysis, and
|
|
investment potential into a single comprehensive report.
|
|
|
|
Args:
|
|
address: The address to analyze (in Hebrew or English)
|
|
years_back: How many years back to analyze (default: 2)
|
|
radius_meters: Search radius in meters (default: 100)
|
|
|
|
Returns:
|
|
JSON string containing:
|
|
- Market activity score and trends
|
|
- Market liquidity and velocity metrics
|
|
- Investment potential analysis
|
|
- Price appreciation and volatility
|
|
"""
|
|
log_mcp_call(
|
|
"get_market_activity_metrics",
|
|
address=address,
|
|
years_back=years_back,
|
|
radius_meters=radius_meters,
|
|
)
|
|
try:
|
|
# Get coordinates for search center
|
|
autocomplete_result = client.autocomplete_address(address)
|
|
search_coords = None
|
|
if autocomplete_result.results:
|
|
coords = autocomplete_result.results[0].coordinates
|
|
if coords:
|
|
search_coords = {"longitude": coords.longitude, "latitude": coords.latitude}
|
|
|
|
# Get deals for the address
|
|
deals = client.find_recent_deals_for_address(address, years_back, radius_meters)
|
|
|
|
analysis_params_base = {
|
|
"address": address,
|
|
"years_back": years_back,
|
|
"radius_meters": radius_meters,
|
|
}
|
|
if search_coords:
|
|
analysis_params_base["search_coordinates"] = search_coords
|
|
|
|
if not deals:
|
|
return json.dumps(
|
|
{
|
|
"analysis_parameters": analysis_params_base,
|
|
"market_statistics": {
|
|
"deal_breakdown": {
|
|
"total_deals": 0,
|
|
},
|
|
},
|
|
"deals": [],
|
|
"error": "No deals found for analysis",
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
# Calculate market metrics using helper to reduce duplication
|
|
activity_metrics = _safe_calculate_metric(client.calculate_market_activity_score, deals)
|
|
liquidity_metrics = _safe_calculate_metric(client.get_market_liquidity, deals)
|
|
investment_metrics = _safe_calculate_metric(client.analyze_investment_potential, deals)
|
|
|
|
# Combine all metrics with normalized structure
|
|
return json.dumps(
|
|
{
|
|
"analysis_parameters": analysis_params_base,
|
|
"market_statistics": {
|
|
"deal_breakdown": {
|
|
"total_deals": len(deals),
|
|
},
|
|
},
|
|
"market_activity": activity_metrics,
|
|
"market_liquidity": liquidity_metrics,
|
|
"investment_potential": investment_metrics,
|
|
"summary": {
|
|
"activity_score": activity_metrics.get("activity_score"),
|
|
"activity_trend": activity_metrics.get("trend"),
|
|
"liquidity_score": liquidity_metrics.get("liquidity_score"),
|
|
"market_activity_level": liquidity_metrics.get("market_activity_level"),
|
|
"investment_score": investment_metrics.get("investment_score"),
|
|
"price_trend": investment_metrics.get("price_trend"),
|
|
"market_stability": investment_metrics.get("market_stability"),
|
|
},
|
|
"deals": [], # Activity metrics don't return raw deals
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error in get_market_activity_metrics: {e}", exc_info=True)
|
|
return f"Error analyzing market activity: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_search_decisive_appraisals_enabled")
|
|
def search_decisive_appraisals(
|
|
block: Optional[str] = None,
|
|
plot: Optional[str] = None,
|
|
decisive_appraiser: Optional[str] = None,
|
|
committee: Optional[str] = None,
|
|
decision_date_from: Optional[str] = None,
|
|
decision_date_to: Optional[str] = None,
|
|
publicity_date_from: Optional[str] = None,
|
|
publicity_date_to: Optional[str] = None,
|
|
search_text: Optional[str] = None,
|
|
appraisal_header: Optional[str] = None,
|
|
max_results: int = 30,
|
|
) -> str:
|
|
"""Search published "decisive appraiser" (שמאי מכריע) decisions from the Israeli
|
|
Ministry of Justice public registry. The registry holds 30,000+ written
|
|
decisions on land betterment levy, expropriation, real-estate disputes etc.
|
|
|
|
All filters are optional. Use block + plot for property-level lookups
|
|
(גוש + חלקה), or `decisive_appraiser` to find every decision by a given
|
|
appraiser. Page size is fixed by the upstream at 10; this tool pages
|
|
automatically up to `max_results`.
|
|
|
|
Args:
|
|
block: Land block number, גוש (e.g. "6212"). Exact match.
|
|
plot: Land plot number, חלקה (e.g. "894"). Use with block.
|
|
decisive_appraiser: Appraiser name in Hebrew (e.g. "דדון דוד").
|
|
committee: Local planning committee (e.g. "תל אביב-יפו").
|
|
decision_date_from / decision_date_to: Date range for the decision date.
|
|
Format: dd-MM-yyyy (e.g. "01-01-2024"). Either bound is optional.
|
|
publicity_date_from / publicity_date_to: Date range for publication.
|
|
search_text: Free-text search inside the decision document.
|
|
appraisal_header: Search within the decision title.
|
|
max_results: Maximum number of decisions to return (default 30).
|
|
|
|
Returns:
|
|
JSON string with `total_results` (database-wide), `returned`, the
|
|
applied `filters`, and a `decisions` array. Each decision contains
|
|
the appraiser name, block/plot, committee, dates, the decision title
|
|
and a `pdf_url` to download the original PDF.
|
|
"""
|
|
log_mcp_call(
|
|
"search_decisive_appraisals",
|
|
block=block,
|
|
plot=plot,
|
|
decisive_appraiser=decisive_appraiser,
|
|
committee=committee,
|
|
decision_date_from=decision_date_from,
|
|
decision_date_to=decision_date_to,
|
|
publicity_date_from=publicity_date_from,
|
|
publicity_date_to=publicity_date_to,
|
|
search_text=search_text,
|
|
appraisal_header=appraisal_header,
|
|
max_results=max_results,
|
|
)
|
|
try:
|
|
response = decisive_appraiser_client.search_decisions_paged(
|
|
max_results=max_results,
|
|
block=block,
|
|
plot=plot,
|
|
decisive_appraiser=decisive_appraiser,
|
|
committee=committee,
|
|
decision_date_from=decision_date_from,
|
|
decision_date_to=decision_date_to,
|
|
publicity_date_from=publicity_date_from,
|
|
publicity_date_to=publicity_date_to,
|
|
search_text=search_text,
|
|
appraisal_header=appraisal_header,
|
|
)
|
|
|
|
decisions = []
|
|
for idx, decision in enumerate(response.results, start=1):
|
|
data = decision.model_dump(mode="json", exclude_none=True)
|
|
documents = data.pop("documents", []) or []
|
|
primary_pdf = documents[0]["file_url"] if documents else None
|
|
data["id"] = idx
|
|
data["pdf_url"] = primary_pdf
|
|
data["all_documents"] = documents
|
|
decisions.append(data)
|
|
|
|
return json.dumps(
|
|
{
|
|
"total_results": response.total_results,
|
|
"returned": len(decisions),
|
|
"filters": {
|
|
"block": block,
|
|
"plot": plot,
|
|
"decisive_appraiser": decisive_appraiser,
|
|
"committee": committee,
|
|
"decision_date_from": decision_date_from,
|
|
"decision_date_to": decision_date_to,
|
|
"publicity_date_from": publicity_date_from,
|
|
"publicity_date_to": publicity_date_to,
|
|
"search_text": search_text,
|
|
"appraisal_header": appraisal_header,
|
|
},
|
|
"decisions": decisions,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=None,
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Error in search_decisive_appraisals: {e}", exc_info=True)
|
|
return f"Error searching decisive appraisals: {str(e)}"
|
|
|
|
|
|
@conditional_tool("tool_download_decisive_appraisal_pdf_enabled")
|
|
def download_decisive_appraisal_pdf(
|
|
url: str,
|
|
filename: Optional[str] = None,
|
|
output_dir: Optional[str] = None,
|
|
overwrite: bool = False,
|
|
) -> str:
|
|
"""Download the PDF of a decisive-appraiser (שמאי מכריע) decision.
|
|
|
|
The PDF host requires the same `x-client-id` header as the search API,
|
|
so a normal browser click on the URL will fail. This tool carries the
|
|
auth header automatically.
|
|
|
|
Args:
|
|
url: Direct PDF URL, as returned in `decisions[].pdf_url` from
|
|
search_decisive_appraisals. Must be on free-justice.openapi.gov.il.
|
|
filename: Optional filename for the saved PDF. Path components are
|
|
stripped; only the basename is used. If omitted, a hash-based
|
|
filename is generated from the URL.
|
|
output_dir: Optional override for the output directory. Defaults to
|
|
the configured `DECISIVE_APPRAISER_DOWNLOAD_DIR`
|
|
(./downloads/decisive_appraisals).
|
|
overwrite: If False (default), returns the cached file when it
|
|
already exists at the destination.
|
|
|
|
Returns:
|
|
JSON string with `path` (absolute local file path), `size_bytes`,
|
|
`from_cache` (true if file already existed), `content_type`.
|
|
"""
|
|
log_mcp_call(
|
|
"download_decisive_appraisal_pdf",
|
|
url=url,
|
|
filename=filename,
|
|
output_dir=output_dir,
|
|
overwrite=overwrite,
|
|
)
|
|
try:
|
|
result = decisive_appraiser_client.download_pdf(
|
|
url=url,
|
|
output_dir=output_dir,
|
|
filename=filename,
|
|
overwrite=overwrite,
|
|
)
|
|
return json.dumps(result, ensure_ascii=False, indent=None)
|
|
except Exception as e:
|
|
logger.error(f"Error in download_decisive_appraisal_pdf: {e}", exc_info=True)
|
|
return f"Error downloading appraisal PDF: {str(e)}"
|
|
|
|
|
|
# Health check endpoint for HTTP deployments
|
|
@mcp.custom_route("/health", methods=["GET"])
|
|
async def health_check(request):
|
|
"""
|
|
Health check endpoint for container orchestration platforms (Render, Railway, etc.).
|
|
|
|
Returns a simple OK status to indicate the server is running.
|
|
This endpoint is accessible at GET /health when using HTTP transport.
|
|
"""
|
|
return JSONResponse({"status": "ok", "service": "nadlan-mcp"})
|
|
|
|
|
|
# Run the server
|
|
if __name__ == "__main__":
|
|
mcp.run()
|