initial

finn_eiendom/mcp_server.py

@@ -0,0 +1,160 @@
+"""FastMCP stdio server for FINN real estate analysis and Eiendom.no enrichment."""
+
+import json
+import logging
+
+from mcp.server.fastmcp import FastMCP
+
+from .analysis import analyze_search
+from .eiendom_no import (
+    build_unit_vector,
+    decode_unit_vector,
+    get_similar_units,
+    get_unit,
+    search_unit_from_finn_url,
+)
+from .service import get_or_fetch_ad, get_or_fetch_eiendom_unit
+
+logger = logging.getLogger(__name__)
+
+mcp = FastMCP("finn_eiendom_mcp")
+
+
+@mcp.tool(
+    description=(
+        "Analyze a FINN.no real estate search URL. Scrapes listing cards,"
+        " fetches details, enriches with Eiendom.no data, scores, and ranks."
+    )
+)
+async def finn_analyze_search(
+    search_url: str,
+    max_pages: int = 3,
+    detail_limit: int = 20,
+    include_details: bool = True,
+    include_eiendom_no: bool = True,
+) -> str:
+    """Analyze a FINN search URL and return ranked listing results."""
+    try:
+        result = await analyze_search(
+            search_url,
+            max_pages=max_pages,
+            fetch_details=include_details,
+            detail_limit=detail_limit,
+            include_eiendom_no=include_eiendom_no,
+        )
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Error analyzing search: {e}")
+        return json.dumps({"error": True, "message": str(e)})
+
+
+@mcp.tool(
+    description=(
+        "Fetch full detail for a FINN listing by finnkode."
+        " Checks cache first; use force_refresh=True to bypass."
+    )
+)
+async def finn_get_ad(finnkode: str, force_refresh: bool = False) -> str:
+    """Fetch FINN ad details by finnkode."""
+    try:
+        ad = await get_or_fetch_ad(finnkode, force_refresh=force_refresh)
+        return ad.model_dump_json()
+    except Exception as e:
+        logger.error(f"Error fetching ad {finnkode}: {e}")
+        return json.dumps({"error": True, "message": str(e)})
+
+
+@mcp.tool(
+    description="Resolve an Eiendom.no unit_code from a FINN listing URL. "
+    "Returns unit_code, address, lat, lng or an error if not found."
+)
+async def finn_resolve_eiendom_unit(finn_url: str) -> str:
+    """Resolve Eiendom.no unit from FINN URL."""
+    try:
+        unit = await search_unit_from_finn_url(finn_url)
+        if unit is None:
+            return json.dumps(
+                {
+                    "error": True,
+                    "message": "Eiendom.no unit could not be resolved from FINN URL",
+                }
+            )
+        return json.dumps(
+            {
+                "unit_code": unit.unit_code,
+                "address": unit.address,
+                "lat": unit.lat,
+                "lng": unit.lng,
+            }
+        )
+    except Exception as e:
+        logger.error(f"Error resolving unit from {finn_url}: {e}")
+        return json.dumps({"error": True, "message": str(e)})
+
+
+@mcp.tool(
+    description="Fetch full Eiendom.no unit data by unit_code. Checks SQLite cache (24h TTL)."
+)
+async def finn_get_eiendom_unit(unit_code: str, force_refresh: bool = False) -> str:
+    """Fetch Eiendom.no unit details by unit_code."""
+    try:
+        unit = await get_or_fetch_eiendom_unit(unit_code, force_refresh=force_refresh)
+        if unit is None:
+            return json.dumps({"error": True, "message": "Eiendom.no unit not found"})
+        return unit.model_dump_json()
+    except Exception as e:
+        logger.error(f"Error fetching unit {unit_code}: {e}")
+        return json.dumps({"error": True, "message": str(e)})
+
+
+@mcp.tool(
+    description="Fetch comparable recently-sold or for-sale units from Eiendom.no using a "
+    "base64-encoded unit vector. Returns list of similar units with sale prices."
+)
+async def finn_get_similar_units(unit_vector: str, listing_status: str = "RECENTLY_SOLD") -> str:
+    """Fetch similar units from Eiendom.no."""
+    try:
+        units = await get_similar_units(unit_vector, listing_status)
+        return json.dumps([unit.model_dump() for unit in units])
+    except Exception as e:
+        logger.error(f"Error fetching similar units: {e}")
+        return json.dumps({"error": True, "message": str(e)})
+
+
+@mcp.tool(
+    description="Build a base64-encoded unit vector for a given Eiendom.no unit_code. "
+    "The vector is used as input to finn_get_similar_units."
+)
+async def finn_build_unit_vector(unit_code: str) -> str:
+    """Build unit vector for Eiendom.no unit."""
+    try:
+        unit = await get_unit(unit_code)
+        if unit is None:
+            return json.dumps({"error": True, "message": "Eiendom.no unit not found"})
+        return json.dumps({"unit_code": unit.unit_code, "unit_vector": build_unit_vector(unit)})
+    except Exception as e:
+        logger.error(f"Error building unit vector for {unit_code}: {e}")
+        return json.dumps({"error": True, "message": str(e)})
+
+
+@mcp.tool(
+    description="Decode a base64 unit vector into human-readable JSON (lat, lon, property type, "
+    "floor, rooms, construction year, area, price)."
+)
+def finn_decode_unit_vector(unit_vector: str) -> str:
+    """Decode unit vector to readable format."""
+    try:
+        result = decode_unit_vector(unit_vector)
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Error decoding unit vector: {e}")
+        return json.dumps({"error": True, "message": str(e)})
+
+
+def main() -> None:
+    """Run the FastMCP stdio server."""
+    mcp.run(transport="stdio")
+
+
+if __name__ == "__main__":
+    main()