mcp_server.py

#!/usr/bin/env python3
"""
kleinanzeigen-reader — MCP server (works with ANY MCP-capable AI system:
Claude Desktop/Code, ChatGPT desktop, Cursor, Cline, Windsurf, Zed, …).

Exposes the reader as Model Context Protocol tools over stdio.

Install + run:
    pip install "kleinanzeigen-reader[mcp]"
    kleinanzeigen-mcp                          # or: python mcp_server.py

Register (example client config, e.g. Claude Desktop / Cursor mcp.json):
    {
      "mcpServers": {
        "kleinanzeigen": { "command": "kleinanzeigen-mcp" }
      }
    }

The core library stays zero-dependency; only this optional server needs the `mcp`
package. Everything here is read-only public data (see references/mobile-api.md).
"""
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

try:
    from mcp.server.fastmcp import FastMCP
except ImportError:  # pragma: no cover
    sys.stderr.write(
        "The 'mcp' package is required for the MCP server.\n"
        "Install it with:  pip install \"kleinanzeigen-reader[mcp]\"\n"
    )
    sys.exit(1)

from kl_reader import (
    fetch, extract, get_flags, get_rating,
    view_counter, seller_other_ads, price_reduction, detect_status, PriceTracker,
)
from kl_reader.comparator import is_duplicate

mcp = FastMCP("kleinanzeigen-reader")


def _analyze(url: str) -> dict:
    r = fetch(url)
    if not r.get("success"):
        return {"error": r.get("error", "fetch failed"), "url": url}
    data = extract(r["html"])
    data.pop("_attrs", None)
    flags = get_flags(data)
    data["flags"] = [{"level": lvl, "message": msg} for lvl, msg in flags]
    data["rating"] = get_rating(data, flags)
    data["status"] = detect_status(r["html"])
    red = price_reduction(r["html"])
    if red:
        data["price_reduction"] = red
    data["seller_other_ads"] = seller_other_ads(r["html"])[:20]
    return data


@mcp.tool()
def read_listing(url: str) -> dict:
    """Analyze a single kleinanzeigen.de listing. Returns structured data:
    title, price, location, category, all attributes, seller type (private/dealer),
    red flags, a 1-5 rating, status (active/reserved/sold), price reduction, and image URLs.
    Use this whenever a user shares a kleinanzeigen.de/s-anzeige/... link."""
    return _analyze(url)


@mcp.tool()
def compare_listings(urls: list) -> dict:
    """Compare two or more kleinanzeigen.de listings side by side and detect whether
    they are likely the same item posted twice. Pass a list of listing URLs."""
    items = [_analyze(u) for u in urls]
    result = {"listings": items}
    if len(items) >= 2 and "error" not in items[0] and "error" not in items[1]:
        result["duplicate_check"] = is_duplicate(items[0], items[1])
    return result


@mcp.tool()
def listing_views(url: str) -> dict:
    """Get the public view counter (numVisits) of a listing. Many views + an old
    posting date usually means the item is overpriced or stale — negotiation leverage."""
    return {"url": url, "views": view_counter(url, referer=url)}


@mcp.tool()
def seller_other_listings(url: str) -> dict:
    """List the seller's other active listings (works for private sellers too) —
    useful to spot undeclared dealers and to gather local price comparisons."""
    r = fetch(url)
    if not r.get("success"):
        return {"error": r.get("error", "fetch failed"), "url": url}
    return {"url": url, "other_ads": seller_other_ads(r["html"])}


@mcp.tool()
def track_price(url: str) -> dict:
    """Record the current price of a listing into a local history store and return the
    full price series over time. Call repeatedly (e.g. daily) to build a real price curve
    that Kleinanzeigen itself does not expose."""
    data = _analyze(url)
    if "error" in data:
        return data
    price = data.get("price_exact") or data.get("price_raw")
    aid = data.get("id")
    pt = PriceTracker()
    if aid and price:
        pt.record(aid, price, url=url)
    return {"url": url, "id": aid, "current_price": price, "history": pt.series(aid) if aid else []}


def main():
    """Console-script entry point (kleinanzeigen-mcp)."""
    mcp.run()


if __name__ == "__main__":
    main()