mcp_server.py

"""
Property Comps MCP Server
Enables Claude, ChatGPT, and other AI agents to look up comparable property sales
across 11 global markets via the Model Context Protocol.

Install: pip install mcp httpx
Run:     python mcp_server.py
Config:  Add to Claude Desktop's claude_desktop_config.json or Claude Code's settings

Example queries an AI agent can answer:
- "What are comparable sales near 10001 in NYC?"
- "What's the median price in Dubai Marina?"
- "Show me price trends for SW1A in London"
- "Generate a PDF valuation report for postcode 75001 in Paris"
"""

import json
import sys
import os

try:
    from mcp.server import Server
    from mcp.server.stdio import stdio_server
    from mcp.types import Tool, TextContent
except ImportError:
    print("Installing MCP SDK...")
    import subprocess
    subprocess.check_call([sys.executable, "-m", "pip", "install", "mcp", "httpx", "-q"])
    from mcp.server import Server
    from mcp.server.stdio import stdio_server
    from mcp.types import Tool, TextContent

import httpx

# ── Market Configuration ──

MARKETS = {
    "uk": {
        "name": "United Kingdom",
        "base_url": "https://property.nwc-advisory.com",
        "location_param": "postcode",
        "location_example": "SW1A1AA",
        "radius_param": "radius_miles",
        "currency": "GBP",
        "transactions": "31M",
    },
    "fr": {
        "name": "France",
        "base_url": "https://property-fr.nwc-advisory.com",
        "location_param": "code_postal",
        "location_example": "75001",
        "radius_param": "radius_km",
        "currency": "EUR",
        "transactions": "8.3M",
    },
    "sg": {
        "name": "Singapore",
        "base_url": "https://property-sg.nwc-advisory.com",
        "location_param": "postal_code",
        "location_example": "310093",
        "radius_param": "radius_km",
        "currency": "SGD",
        "transactions": "973K",
    },
    "nyc": {
        "name": "New York City",
        "base_url": "https://property-nyc.nwc-advisory.com",
        "location_param": "zip_code",
        "location_example": "10001",
        "radius_param": "radius_miles",
        "currency": "USD",
        "transactions": "505K",
    },
    "chi": {
        "name": "Chicago",
        "base_url": "https://property-chi.nwc-advisory.com",
        "location_param": "zip_code",
        "location_example": "60606",
        "radius_param": "radius_miles",
        "currency": "USD",
        "transactions": "282K",
    },
    "dxb": {
        "name": "Dubai",
        "base_url": "https://property-dxb.nwc-advisory.com",
        "location_param": "area_name",
        "location_example": "Dubai Marina",
        "radius_param": "radius_km",
        "currency": "AED",
        "transactions": "1M",
    },
    "mia": {
        "name": "Miami",
        "base_url": "https://property-miami.nwc-advisory.com",
        "location_param": "zip_code",
        "location_example": "33101",
        "radius_param": "radius_miles",
        "currency": "USD",
        "transactions": "284K",
    },
    "phl": {
        "name": "Philadelphia",
        "base_url": "https://property-phl.nwc-advisory.com",
        "location_param": "zip_code",
        "location_example": "19103",
        "radius_param": "radius_miles",
        "currency": "USD",
        "transactions": "240K",
    },
    "ct": {
        "name": "Connecticut",
        "base_url": "https://property-ct.nwc-advisory.com",
        "location_param": "town",
        "location_example": "Greenwich",
        "radius_param": "radius_miles",
        "currency": "USD",
        "transactions": "282K",
    },
    "ie": {
        "name": "Ireland",
        "base_url": "https://property-ie.nwc-advisory.com",
        "location_param": "county",
        "location_example": "Dublin",
        "radius_param": "radius_km",
        "currency": "EUR",
        "transactions": "230K",
    },
    "tw": {
        "name": "Taiwan",
        "base_url": "https://property-tw.nwc-advisory.com",
        "location_param": "city_en",
        "location_example": "Taipei",
        "radius_param": "radius_km",
        "currency": "TWD",
        "transactions": "669K",
    },
}


def detect_market(location: str) -> str:
    """Auto-detect market from location format."""
    loc = location.strip()

    # UK postcode pattern (letters+digits+space+digit+letters)
    if len(loc) >= 5 and loc[0].isalpha() and any(c.isdigit() for c in loc[:4]):
        if " " in loc or len(loc) <= 8:
            return "uk"

    # Singapore (6 digits)
    if len(loc) == 6 and loc.isdigit():
        return "sg"

    # 5-digit codes: distinguish US ZIP from French postal code
    if len(loc) == 5 and loc.isdigit():
        first2 = int(loc[:2])
        first3 = loc[:3]

        # US ZIP prefixes (check these first -- they overlap with French 01-95)
        us_nyc = ("100", "101", "102", "103", "104", "110", "111", "112", "113", "114", "116")
        us_mia = ("330", "331", "332", "333", "334", "341")
        us_phl = ("190", "191", "192", "193", "194")
        us_chi = ("604", "605", "606", "607", "608")
        us_ct = ("060", "061", "062", "063", "064", "065", "066", "067", "068", "069")

        if first3 in us_nyc:
            return "nyc"
        if first3 in us_mia:
            return "mia"
        if first3 in us_phl:
            return "phl"
        if first3 in us_chi:
            return "chi"
        if first3 in us_ct:
            return "ct"

        # Other US ZIPs (ranges not covered by France: 96, 99, and some others)
        # French codes: 01-95, 97, 98. US ZIPs: 00-99 (overlap is total for 01-95)
        # Heuristic: if first2 > 95 or first2 == 0, it's US
        if first2 == 0 or first2 > 95:
            return "nyc"  # default US

        # Remaining 5-digit codes (01-95): assume French
        return "fr"

    # Named locations
    loc_lower = loc.lower()
    if any(w in loc_lower for w in ["dubai", "marina", "jumeirah", "palm", "jbr", "downtown"]):
        return "dxb"
    if any(w in loc_lower for w in ["greenwich", "stamford", "hartford", "new haven", "connecticut"]):
        return "ct"
    if any(w in loc_lower for w in ["dublin", "cork", "galway", "limerick"]):
        return "ie"
    if any(w in loc_lower for w in ["taipei", "kaohsiung", "taichung", "tainan"]):
        return "tw"
    if any(w in loc_lower for w in ["bukit", "ang mo", "tampines", "woodlands", "jurong", "toa payoh"]):
        return "sg"

    return "uk"  # fallback


# ── MCP Server ──

server = Server("property-comps")


@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="search_property_comps",
            description=(
                "Search for comparable property sales near a location. "
                "Covers 11 markets: NYC, Miami, Philadelphia, Connecticut, Chicago, "
                "Dubai, Singapore, London/UK, Paris/France, Ireland, Taiwan. "
                "Returns recent sales with price, date, address, property type, and area. "
                "Also returns statistics (median, average, min, max)."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "Postcode, ZIP code, or area name. Examples: SW1A1AA (UK), 10001 (NYC), 75001 (Paris), Dubai Marina, 310093 (Singapore)",
                    },
                    "market": {
                        "type": "string",
                        "description": "Market code (optional, auto-detected from location). One of: uk, fr, sg, nyc, chi, dxb, mia, phl, ct, ie, tw",
                        "enum": list(MARKETS.keys()),
                    },
                    "months": {
                        "type": "integer",
                        "description": "Look-back period in months (default: 12)",
                        "default": 12,
                    },
                    "radius": {
                        "type": "number",
                        "description": "Search radius in miles (UK/US) or km (others). Default: 1.0",
                        "default": 1.0,
                    },
                    "property_type": {
                        "type": "string",
                        "description": "Filter by type. UK: D/S/T/F. NYC: Condo/1-Family. SG: 3 ROOM/4 ROOM/5 ROOM. Optional.",
                    },
                    "limit": {
                        "type": "integer",
                        "description": "Max results (default: 10)",
                        "default": 10,
                    },
                },
                "required": ["location"],
            },
        ),
        Tool(
            name="get_area_stats",
            description=(
                "Get price statistics for a property market area. "
                "Returns median, average, min, max prices with breakdown by property type. "
                "Covers all 11 markets."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "Postcode, ZIP code, or area name",
                    },
                    "market": {
                        "type": "string",
                        "enum": list(MARKETS.keys()),
                    },
                    "months": {
                        "type": "integer",
                        "default": 12,
                    },
                },
                "required": ["location"],
            },
        ),
        Tool(
            name="list_markets",
            description="List all 11 available property markets with transaction counts and examples.",
            inputSchema={
                "type": "object",
                "properties": {},
            },
        ),
    ]


@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "list_markets":
        lines = ["Available property markets:\n"]
        for key, m in MARKETS.items():
            lines.append(
                f"- **{m['name']}** ({key}): {m['transactions']} transactions, "
                f"currency {m['currency']}, example: {m['location_param']}={m['location_example']}"
            )
        lines.append(f"\nTotal: 4.2M+ transactions across 11 markets.")
        lines.append("Data from official government registries.")
        return [TextContent(type="text", text="\n".join(lines))]

    location = arguments.get("location", "")
    market_key = arguments.get("market") or detect_market(location)
    market = MARKETS.get(market_key)

    if not market:
        return [TextContent(type="text", text=f"Unknown market: {market_key}. Use list_markets to see available options.")]

    months = arguments.get("months", 12)
    radius = arguments.get("radius", 1.0)
    limit = arguments.get("limit", 10)

    params = {
        market["location_param"]: location,
        market["radius_param"]: radius,
        "months": months,
        "limit": limit,
    }

    if arguments.get("property_type"):
        params["property_type"] = arguments["property_type"]

    if name == "search_property_comps":
        endpoint = "/v1/comps"
    elif name == "get_area_stats":
        endpoint = "/v1/stats"
    else:
        return [TextContent(type="text", text=f"Unknown tool: {name}")]

    url = f"{market['base_url']}{endpoint}"

    try:
        async with httpx.AsyncClient(timeout=30) as client:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
            data = resp.json()
    except httpx.HTTPStatusError as e:
        return [TextContent(type="text", text=f"API error ({e.response.status_code}): {e.response.text[:500]}")]
    except Exception as e:
        return [TextContent(type="text", text=f"Connection error to {market['name']}: {str(e)}")]

    # Format response
    if name == "search_property_comps":
        results = data.get("results", [])
        stats = data.get("statistics", {})
        count = data.get("count", len(results))

        lines = [f"**{market['name']}** - {count} comparable sales found near {location}\n"]

        if stats:
            lines.append(f"**Statistics:** Median: {market['currency']} {stats.get('median', 'N/A'):,} | "
                        f"Average: {market['currency']} {stats.get('average', 'N/A'):,} | "
                        f"Range: {market['currency']} {stats.get('min', 'N/A'):,} - {market['currency']} {stats.get('max', 'N/A'):,}\n")

        for i, r in enumerate(results[:limit], 1):
            price = r.get("price") or r.get("sale_price") or r.get("resale_price", 0)
            date = r.get("date") or r.get("sale_date", "")
            address = r.get("address") or r.get("street_name", "")
            ptype = r.get("property_type") or r.get("flat_type", "")
            dist = r.get("distance_miles") or r.get("distance_km", "")
            area = r.get("floor_area_sqm") or r.get("gross_sqft", "")

            lines.append(f"{i}. {market['currency']} {price:,} | {date} | {ptype} | {address}")
            if dist:
                lines.append(f"   Distance: {dist} | Area: {area}")

        lines.append(f"\nData source: {market['name']} government property registry")
        return [TextContent(type="text", text="\n".join(lines))]

    elif name == "get_area_stats":
        stats = data.get("stats", data.get("statistics", {}))
        by_type = data.get("by_type", {})

        lines = [f"**{market['name']}** - Area statistics for {location}\n"]

        if stats:
            lines.append(f"Median: {market['currency']} {stats.get('median', 'N/A'):,}")
            lines.append(f"Average: {market['currency']} {stats.get('average', 'N/A'):,}")
            lines.append(f"Min: {market['currency']} {stats.get('min', 'N/A'):,}")
            lines.append(f"Max: {market['currency']} {stats.get('max', 'N/A'):,}")
            lines.append(f"Transactions: {stats.get('count', 'N/A')}")

        if by_type:
            lines.append(f"\n**By property type:**")
            for ptype, pstats in by_type.items():
                lines.append(f"- {ptype}: Median {market['currency']} {pstats.get('median', 'N/A'):,} ({pstats.get('count', 0)} sales)")

        return [TextContent(type="text", text="\n".join(lines))]


async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())