csfloat_mcp_server.py

#!/usr/bin/env python3
"""CSFloat MCP server.

Hard rule: nothing in this process may write to stdout. The MCP stdio transport
uses stdout as the JSON-RPC channel, so a stray print() corrupts the stream and
the client drops the session with an unrelated-looking parse error. Diagnostics
go to stderr through `log`.

All deterministic domain logic (wear ranges, weapon resolution, name shaping,
deal math) lives in csfloat_domain.py so it can be tested without a server.
"""

from __future__ import annotations

import asyncio
import logging
import os
import sys
from typing import Any, Literal, Optional, get_args

import httpx
from mcp.server.fastmcp import FastMCP

from csfloat_domain import (
    CATEGORY_ANY,
    NAME_FORMAT_HINT,
    WEAPONS_LEFT_OUT,
    WEAR_ALIASES,
    WEAR_CAP_HINT,
    WEAR_RANGES,
    DealMetrics,
    WeaponResolutionError,
    api_float_bounds,
    deal_label,
    deal_metrics,
    get_wear_conditions,
    get_weapon_def_index_mapping,
    normalize_market_hash_name,
    parse_query,
    resolve_weapon,
    to_cents,
)

log = logging.getLogger("csfloat_mcp")

mcp = FastMCP("csfloat")

CSFLOAT_API_BASE = "https://csfloat.com/api/v1"
USER_AGENT = "CSFloat-MCP-Server/1.0"
MAX_LIMIT = 50

SortBy = Literal[
    "best_deal",
    "lowest_price",
    "highest_price",
    "lowest_float",
    "highest_float",
    "most_recent",
    "highest_discount",
    "expires_soon",
    "float_rank",
    "num_bids",
]
SORT_OPTIONS = frozenset(get_args(SortBy))

MISSING_KEY_MESSAGE = (
    "CSFLOAT_API_KEY is not set in the MCP server environment. Ask the operator to put "
    "the key in the `env` block of the MCP client config (a shell `export` does not reach "
    "GUI clients like Claude Desktop or Cursor). Get a key at https://csfloat.com/profile. "
    "This is a configuration problem, not an empty market."
)


class CSFloatError(Exception):
    """A CSFloat request failed for a reason the model should hear verbatim."""


# --------------------------------------------------------------------------- #
# HTTP
# --------------------------------------------------------------------------- #

_client: Optional[httpx.AsyncClient] = None


def _http() -> httpx.AsyncClient:
    """One client for the process: keep-alive instead of a TLS handshake per tool call."""
    global _client
    if _client is None:
        _client = httpx.AsyncClient(
            timeout=httpx.Timeout(10.0, read=25.0),
            headers={"User-Agent": USER_AGENT, "Accept": "application/json"},
        )
    return _client


def _api_key() -> str:
    return (os.getenv("CSFLOAT_API_KEY") or "").strip()


def _retry_after(response: httpx.Response, default: float = 2.0) -> float:
    try:
        return max(0.0, float(response.headers.get("Retry-After", default)))
    except (TypeError, ValueError):
        return default


def _payload(response: httpx.Response) -> dict[str, Any]:
    """Validate the response shape and surface a 200-with-error-body once, here."""
    try:
        body = response.json()
    except ValueError as exc:
        raise CSFloatError("CSFloat returned a non-JSON response.") from exc
    if isinstance(body, list):
        return {"data": body}
    if not isinstance(body, dict):
        raise CSFloatError(f"Unexpected CSFloat response type: {type(body).__name__}.")
    if "data" not in body and body.get("message"):
        raise CSFloatError(f"CSFloat API error: {body['message']}")
    return body


async def make_csfloat_request(url: str, params: Optional[dict] = None) -> dict[str, Any]:
    """GET the CSFloat API. Returns the parsed body or raises CSFloatError.

    Never returns None: 'request failed' and 'zero results' are opposite
    conclusions and must not collapse into the same value.
    """
    key = _api_key()
    if not key:
        raise CSFloatError(MISSING_KEY_MESSAGE)

    headers = {"Authorization": key}
    pending: Optional[CSFloatError] = None
    delay = 1.0

    for attempt in (0, 1):
        try:
            response = await _http().get(url, headers=headers, params=params or {})
        except httpx.TimeoutException as exc:
            log.warning("CSFloat timeout on %s: %s", url, exc)
            pending = CSFloatError("CSFloat did not respond in time (transient, try again).")
        except httpx.RequestError as exc:
            log.warning("CSFloat request error on %s: %s", url, exc)
            pending = CSFloatError(f"Could not reach CSFloat ({type(exc).__name__}).")
        else:
            status = response.status_code
            if status == 429:
                delay = _retry_after(response)
                log.warning("CSFloat rate limited (retry-after=%ss)", delay)
                pending = CSFloatError(
                    f"CSFloat rate limit (HTTP 429). Wait about {delay:.0f}s before trying "
                    "again and do not repeat or reword the search in the meantime."
                )
            elif status in (401, 403):
                log.error("CSFloat rejected the API key (HTTP %s)", status)
                raise CSFloatError(
                    f"CSFloat rejected the API key (HTTP {status}). It is missing, expired or "
                    "lacks permission. Ask the operator to check CSFLOAT_API_KEY."
                )
            elif status == 404:
                raise CSFloatError("CSFloat returned HTTP 404: that resource does not exist.")
            elif status >= 500:
                log.warning("CSFloat server error (HTTP %s)", status)
                pending = CSFloatError(
                    f"CSFloat server error (HTTP {status}). Transient on their side, try later."
                )
            elif status >= 400:
                detail = (response.text or "").strip()[:200]
                log.warning("CSFloat rejected the request (HTTP %s): %s", status, detail)
                raise CSFloatError(
                    f"CSFloat rejected the request (HTTP {status}): {detail or 'no detail'}. "
                    "Check the parameters, not the skin name."
                )
            else:
                return _payload(response)

        if attempt == 0 and delay <= 5:
            await asyncio.sleep(delay)
        else:
            break

    raise pending or CSFloatError("CSFloat request failed for an unknown reason.")


# --------------------------------------------------------------------------- #
# formatting
# --------------------------------------------------------------------------- #


def _money(value: Optional[float]) -> str:
    return f"${value:.2f}" if value is not None else "unavailable"


def _price_line(metrics: DealMetrics) -> str:
    kind = "AUCTION" if metrics.is_auction else "BUY NOW"
    if metrics.price_usd is None:
        return f"{metrics.price_label}: unavailable ({kind}, no bid yet)"
    return f"{metrics.price_label}: {_money(metrics.price_usd)} ({kind})"


def _predicted_line(metrics: DealMetrics) -> str:
    """Neutral data plus, only when it is defensible, a label."""
    if metrics.pct_vs_predicted is None:
        if metrics.predicted_usd:
            return f"Predicted price: {_money(metrics.predicted_usd)} (not comparable to this price)"
        return ""
    label = deal_label(metrics.pct_vs_predicted, metrics.pattern_sensitive)
    suffix = f" [{label}]" if label else ""
    return (
        f"{metrics.pct_vs_predicted:+.1f}% vs predicted {_money(metrics.predicted_usd)}{suffix}"
    )


def _sp_line(metrics: DealMetrics) -> str:
    stickers = metrics.stickers
    if not stickers.names:
        return ""
    if metrics.sp_percent is None:
        return "SP: unavailable (no sticker reference price)"
    baseline = metrics.base_usd if (metrics.base_usd and stickers.names) else metrics.predicted_usd
    note = ""
    if stickers.worn_ignored:
        note += f", {stickers.worn_ignored} scraped sticker(s) not counted"
    if metrics.keychain_usd:
        note += f", charm value {_money(metrics.keychain_usd)} excluded"
    if metrics.sp_percent == 0:
        note = ", stickers included at no premium over the baseline" + note
    return (
        f"SP: {metrics.sp_percent:.0f}% "
        f"({_money(stickers.total_usd)} sticker value vs {_money(baseline)} baseline{note})"
    )


def _pattern_note(item: dict[str, Any], metrics: DealMetrics) -> str:
    notes = []
    name = (item.get("market_hash_name") or "").lower()
    if metrics.pattern_sensitive:
        notes.append(
            "price here depends on pattern/stickers/charm, which predicted_price does not model"
        )
    if "doppler" in name:
        notes.append("Doppler listings mix Phase 1-4 / Ruby / Sapphire / Black Pearl (>10x spread)")
    return f"Note: {'; '.join(notes)}\n" if notes else ""


def format_listing(listing: dict[str, Any]) -> str:
    """Full detail for one listing."""
    listing = listing or {}
    item = listing.get("item") or {}
    metrics = deal_metrics(listing)
    seller = (listing.get("seller") or {}).get("username") or "Unknown"
    listing_id = listing.get("id")

    lines = [
        f"Item: {item.get('market_hash_name') or 'Unknown'}",
        _price_line(metrics),
        f"Float: {item.get('float_value', 'N/A')} ({item.get('wear_name') or 'N/A'})",
    ]
    if item.get("paint_seed") is not None or item.get("paint_index") is not None:
        lines.append(
            f"Pattern: seed {item.get('paint_seed', 'N/A')} / paint index {item.get('paint_index', 'N/A')}"
        )
    lines += [
        f"Seller: {seller}",
        f"Listing ID: {listing_id or 'N/A'}",
        f"URL: {'https://csfloat.com/item/' + str(listing_id) if listing_id else 'N/A'}",
    ]

    result = "\n".join(lines) + "\n"
    for extra in (_predicted_line(metrics), _sp_line(metrics)):
        if extra:
            result += extra + "\n"

    if metrics.is_auction:
        auction = listing.get("auction_details") or {}
        ends = auction.get("expires_at") or listing.get("expires_at")
        if ends:
            result += f"Auction ends: {ends}\n"

    stickers = metrics.stickers
    if stickers.names:
        parts = []
        for raw in item.get("stickers") or []:
            sticker = raw or {}
            price = ((sticker.get("reference") or {}).get("price"))
            name = sticker.get("name") or "Unknown"
            wear = sticker.get("wear") or 0
            tag = f" ({_money(price / 100)})" if price else ""
            parts.append(f"{name}{tag}{' [scraped]' if wear else ''}")
        result += f"Stickers: {', '.join(parts)}\n"

    if metrics.keychain_usd or (item.get("keychains") or []):
        charms = [
            f"{(kc or {}).get('name') or 'Unknown'}"
            for kc in (item.get("keychains") or [])
        ]
        result += f"Charms: {', '.join(charms)} (total {_money(metrics.keychain_usd)})\n"

    result += _pattern_note(item, metrics)
    return result.strip()


def format_listing_summary(listing: dict[str, Any]) -> str:
    """One line per listing."""
    listing = listing or {}
    item = listing.get("item") or {}
    metrics = deal_metrics(listing)
    listing_id = listing.get("id")
    url = f"https://csfloat.com/item/{listing_id}" if listing_id else "N/A"

    kind = "AUCTION" if metrics.is_auction else "BUY NOW"
    if metrics.price_usd is None:
        price = "price unavailable (no bid yet)" if metrics.is_auction else "price unavailable"
    elif metrics.is_auction:
        price = f"{_money(metrics.price_usd)} {metrics.price_label.lower()}"
    else:
        price = _money(metrics.price_usd)
    delta = ""
    if metrics.pct_vs_predicted is not None:
        label = deal_label(metrics.pct_vs_predicted, metrics.pattern_sensitive)
        delta = f" {metrics.pct_vs_predicted:+.1f}% vs predicted {_money(metrics.predicted_usd)}"
        if label:
            delta += f" [{label}]"
    sp = f" SP:{metrics.sp_percent:.0f}%" if metrics.sp_percent is not None else ""

    return (
        f"{item.get('market_hash_name') or 'Unknown'} - {price} ({kind}){delta}{sp} "
        f"(Float: {item.get('float_value', 'N/A')}) - {url}"
    )


def _safe_summary(listing: Any) -> Optional[str]:
    """One malformed listing must not wipe out the other 19."""
    try:
        return format_listing_summary(listing)
    except Exception:  # noqa: BLE001 - defensive: unknown API shape
        # The row that just failed may not be a Mapping at all, so it cannot be
        # asked for an id: doing so raises out of the handler and loses the page.
        listing_id = listing.get("id") if isinstance(listing, dict) else None
        log.warning("Could not format listing %r", listing_id, exc_info=True)
        return None


# --------------------------------------------------------------------------- #
# shared parameter handling
# --------------------------------------------------------------------------- #


def _validate(
    *,
    limit: int,
    page: int,
    sort_by: str,
    min_price: Optional[float],
    max_price: Optional[float],
    min_float: Optional[float],
    max_float: Optional[float],
) -> tuple[Optional[str], int, int, list[str]]:
    """Returns (error_message, clamped_limit, clamped_page, notes)."""
    if sort_by not in SORT_OPTIONS:
        return (
            f"Invalid sort_by {sort_by!r}. Valid values: {', '.join(sorted(SORT_OPTIONS))}.",
            0,
            0,
            [],
        )
    for name, value in (("min_float", min_float), ("max_float", max_float)):
        if value is not None and not 0.0 <= value <= 1.0:
            return f"{name} must be between 0.0 and 1.0, got {value}.", 0, 0, []
    if min_float is not None and max_float is not None and min_float > max_float:
        return f"min_float ({min_float}) is greater than max_float ({max_float}).", 0, 0, []
    for name, value in (("min_price", min_price), ("max_price", max_price)):
        if value is not None and value < 0:
            return f"{name} cannot be negative, got {value}.", 0, 0, []
    if min_price is not None and max_price is not None and min_price > max_price:
        return f"min_price ({min_price}) is greater than max_price ({max_price}).", 0, 0, []

    notes: list[str] = []
    clamped_limit = max(1, min(int(limit), MAX_LIMIT))
    if clamped_limit != limit:
        notes.append(f"limit adjusted from {limit} to {clamped_limit} (API maximum is {MAX_LIMIT})")
    clamped_page = max(0, int(page))
    if clamped_page != page:
        notes.append(f"page adjusted from {page} to {clamped_page}")
    return None, clamped_limit, clamped_page, notes


def _base_params(
    *,
    limit: int,
    page: int,
    sort_by: str,
    include_auctions: bool,
    min_price: Optional[float],
    max_price: Optional[float],
) -> dict[str, Any]:
    params: dict[str, Any] = {"page": page, "limit": limit, "sort_by": sort_by}
    if not include_auctions:
        params["type"] = "buy_now"
    if min_price is not None:
        params["min_price"] = to_cents(min_price)
    if max_price is not None:
        params["max_price"] = to_cents(max_price)
    return params


async def _run_listing_search(
    *,
    params: dict[str, Any],
    subject: str,
    notes: list[str],
    include_auctions: bool,
    empty_hint: str,
) -> str:
    """Single code path for every /listings query."""
    try:
        data = await make_csfloat_request(f"{CSFLOAT_API_BASE}/listings", params)
    except CSFloatError as exc:
        return f"CSFloat request failed: {exc}"

    rows = data.get("data") or []
    prefix = "".join(f"Note: {note}\n" for note in notes)
    if not rows:
        return (
            f"{prefix}The search for {subject} was valid and returned 0 listings "
            f"(not an error).\n{empty_hint}"
        )

    summaries = [line for line in (_safe_summary(row) for row in rows) if line]
    if not summaries:
        return f"{prefix}Received {len(rows)} listings but none could be formatted (unexpected API shape)."

    header = [
        f"{len(summaries)} listings shown for {subject}",
        f"sorted by {params['sort_by'].replace('_', ' ')}",
        "buy-now only" if not include_auctions else "auctions included",
        f"page {params.get('page', 0)}",
    ]
    footer = ""
    total = data.get("total") or data.get("count")
    if total is not None:
        header.append(f"{total} total available")
    if data.get("cursor"):
        footer = (
            f"\n\nMore results exist. Next page: same call with page="
            f"{params.get('page', 0) + 1} (API cursor: {data['cursor']})."
        )
    elif len(summaries) == params.get("limit"):
        footer = (
            f"\n\nThis is a full page, so more results may exist: repeat with "
            f"page={params.get('page', 0) + 1}."
        )

    body = "\n".join(f"• {line}" for line in summaries)
    return f"{prefix}{' | '.join(header)}:\n\n{body}{footer}"


# --------------------------------------------------------------------------- #
# tools
# --------------------------------------------------------------------------- #


@mcp.tool()
async def search_skins(
    skin_name: str,
    limit: int = 20,
    page: int = 0,
    sort_by: SortBy = "best_deal",
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    min_float: Optional[float] = None,
    max_float: Optional[float] = None,
    include_auctions: bool = False,
) -> str:
    """Search CSFloat for one specific item by its exact market_hash_name.

    Use this when the user named a skin. Use search_by_weapon when they only
    named a weapon ('cheap AWPs'), and browse_market when they named neither.

    market_hash_name is an exact match, so the shape matters:
      '<Weapon> | <Skin> (<Wear>)'  e.g. 'AK-47 | Redline (Field-Tested)'
      knives and gloves need '★ '   e.g. '★ Karambit | Fade (Factory New)'
      StatTrak needs 'StatTrak™ '   e.g. 'StatTrak™ AK-47 | Redline (Field-Tested)'
      Souvenir needs 'Souvenir '    e.g. 'Souvenir AWP | Dragon Lore (Field-Tested)'
    Wear conditions: Factory New (0.00-0.07), Minimal Wear (0.07-0.15),
    Field-Tested (0.15-0.38), Well-Worn (0.38-0.45), Battle-Scarred (0.45-1.00).
    Abbreviations (FN/MW/FT/WW/BS) and lowercase are accepted and normalised.
    Items with no wear (cases, stickers, agents, patches, charms, music kits,
    vanilla knives) are searched by their exact name, with no wear in it.

    Args:
        skin_name: Exact market_hash_name of the item.
        limit: Results to return (1-50).
        page: Page number, 0-based.
        sort_by: Sort order.
        min_price: Minimum price in USD.
        max_price: Maximum price in USD.
        min_float: Minimum float (0.0-1.0).
        max_float: Maximum float (0.0-1.0).
        include_auctions: Include auction listings (default False, buy-now only).
    """
    raw_name = (skin_name or "").strip()
    if not raw_name:
        return f"skin_name is required. {NAME_FORMAT_HINT}"

    error, limit, page, notes = _validate(
        limit=limit,
        page=page,
        sort_by=sort_by,
        min_price=min_price,
        max_price=max_price,
        min_float=min_float,
        max_float=max_float,
    )
    if error:
        return error

    # A bare weapon name means the user wants the whole weapon, not one skin.
    # Only redirect on an exact weapon alias: substring matching used to turn
    # 'AK-47 Redline' into 'every AK-47' and report it as Redline.
    parsed = parse_query(raw_name)
    if "|" not in raw_name and parsed.weapon in get_weapon_def_index_mapping():
        return await _search_by_weapon(
            weapon_name=raw_name,
            limit=limit,
            page=page,
            sort_by=sort_by,
            min_price=min_price,
            max_price=max_price,
            min_float=min_float,
            max_float=max_float,
            include_auctions=include_auctions,
        )

    name = normalize_market_hash_name(raw_name)
    if name != raw_name:
        notes.append(f"searched as {name!r} (CSFloat's exact name format)")

    params = _base_params(
        limit=limit,
        page=page,
        sort_by=sort_by,
        include_auctions=include_auctions,
        min_price=min_price,
        max_price=max_price,
    )
    params["market_hash_name"] = name
    if min_float is not None:
        params["min_float"] = min_float
    if max_float is not None:
        params["max_float"] = max_float

    return await _run_listing_search(
        params=params,
        subject=f"'{name}'",
        notes=notes,
        include_auctions=include_auctions,
        empty_hint=(
            f"Either nobody is selling it right now, or the name does not match exactly. "
            f"{NAME_FORMAT_HINT}"
        ),
    )


@mcp.tool()
async def get_listing_details(listing_id: str) -> str:
    """Get full detail for one listing (price vs predicted, float, pattern, stickers, charms).

    Args:
        listing_id: Numeric listing id, as returned by the search tools.
    """
    clean = (listing_id or "").strip()
    if not clean.isdigit():
        return f"Invalid listing id {listing_id!r}: CSFloat listing ids are numeric."

    try:
        data = await make_csfloat_request(f"{CSFLOAT_API_BASE}/listings/{clean}")
    except CSFloatError as exc:
        return f"CSFloat request failed: {exc}"

    if "id" not in data:
        return f"Listing {clean} not found, or the response had an unexpected shape."
    try:
        return format_listing(data)
    except Exception:  # noqa: BLE001 - defensive: unknown API shape
        log.warning("Could not format listing %s", clean, exc_info=True)
        return f"Listing {clean} was returned but could not be formatted (unexpected API shape)."


@mcp.tool()
async def browse_market(
    page: int = 0,
    limit: int = 20,
    sort_by: SortBy = "best_deal",
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    include_auctions: bool = False,
) -> str:
    """Browse CSFloat listings with no item filter. Use when no weapon or skin was named.

    Args:
        page: Page number, 0-based.
        limit: Results per page (1-50).
        sort_by: Sort order.
        min_price: Minimum price in USD.
        max_price: Maximum price in USD.
        include_auctions: Include auction listings (default False, buy-now only).
    """
    error, limit, page, notes = _validate(
        limit=limit,
        page=page,
        sort_by=sort_by,
        min_price=min_price,
        max_price=max_price,
        min_float=None,
        max_float=None,
    )
    if error:
        return error

    params = _base_params(
        limit=limit,
        page=page,
        sort_by=sort_by,
        include_auctions=include_auctions,
        min_price=min_price,
        max_price=max_price,
    )
    return await _run_listing_search(
        params=params,
        subject="the whole market",
        notes=notes,
        include_auctions=include_auctions,
        empty_hint="Try a wider price range or a different sort order.",
    )


@mcp.tool()
async def search_by_weapon(
    weapon_name: str,
    limit: int = 20,
    page: int = 0,
    sort_by: SortBy = "best_deal",
    min_price: Optional[float] = None,
    max_price: Optional[float] = None,
    min_float: Optional[float] = None,
    max_float: Optional[float] = None,
    include_auctions: bool = False,
) -> str:
    """Search every skin of one weapon, by def_index. Use when the user named a weapon only.

    This cannot filter by skin: for one specific skin use search_skins with the
    exact market_hash_name. A wear condition in the text is turned into a float
    filter, and 'StatTrak'/'Souvenir' into a category filter, so
    'StatTrak AK-47 FT' and '★ Karambit FN' work.

    Args:
        weapon_name: Weapon, optionally with wear/quality (e.g. 'AWP', 'AK-47 FT', 'ST M4A1-S MW').
        limit: Results to return (1-50).
        page: Page number, 0-based.
        sort_by: Sort order.
        min_price: Minimum price in USD.
        max_price: Maximum price in USD.
        min_float: Minimum float (0.0-1.0), overrides the wear condition.
        max_float: Maximum float (0.0-1.0), overrides the wear condition.
        include_auctions: Include auction listings (default False, buy-now only).
    """
    return await _search_by_weapon(
        weapon_name=weapon_name,
        limit=limit,
        page=page,
        sort_by=sort_by,
        min_price=min_price,
        max_price=max_price,
        min_float=min_float,
        max_float=max_float,
        include_auctions=include_auctions,
    )


async def _search_by_weapon(
    *,
    weapon_name: str,
    limit: int,
    page: int,
    sort_by: str,
    min_price: Optional[float],
    max_price: Optional[float],
    min_float: Optional[float],
    max_float: Optional[float],
    include_auctions: bool,
) -> str:
    """Body of search_by_weapon, callable from search_skins without going through the tool."""
    raw = (weapon_name or "").strip()
    if not raw:
        return (
            "weapon_name is required (e.g. 'AWP', 'AK-47 Field-Tested'). "
            f"Items with no def_index: {'; '.join(WEAPONS_LEFT_OUT)}."
        )

    error, limit, page, notes = _validate(
        limit=limit,
        page=page,
        sort_by=sort_by,
        min_price=min_price,
        max_price=max_price,
        min_float=min_float,
        max_float=max_float,
    )
    if error:
        return error

    parsed = parse_query(raw)
    try:
        match = resolve_weapon(parsed.weapon)
    except WeaponResolutionError as exc:
        return (
            f"{exc}\nSearch a specific item with search_skins and its exact market_hash_name "
            f"instead. Items with no def_index: {'; '.join(WEAPONS_LEFT_OUT)}."
        )

    if not match.exact:
        notes.append(f"read {raw!r} as weapon {match.key!r} (def_index {match.def_index})")
    if match.leftover:
        notes.append(
            f"a def_index search cannot filter by skin, so {', '.join(repr(t) for t in match.leftover)} "
            f"was IGNORED - these are ALL {match.key} listings, not that specific skin. "
            "Use search_skins with the full market_hash_name for one skin."
        )

    params = _base_params(
        limit=limit,
        page=page,
        sort_by=sort_by,
        include_auctions=include_auctions,
        min_price=min_price,
        max_price=max_price,
    )
    params["def_index"] = match.def_index
    if parsed.category != CATEGORY_ANY:
        params["category"] = parsed.category
        notes.append(f"category filter {parsed.category} (1=normal, 2=StatTrak, 3=Souvenir)")

    wear_bounds = api_float_bounds(parsed.wear) if parsed.wear else None
    if min_float is not None:
        params["min_float"] = min_float
    elif wear_bounds:
        params["min_float"] = wear_bounds[0]
    if max_float is not None:
        params["max_float"] = max_float
    elif wear_bounds:
        params["max_float"] = wear_bounds[1]

    subject = f"weapon {match.key!r} (def_index {match.def_index})"
    if parsed.wear:
        low, high = WEAR_RANGES[parsed.wear]
        subject += f" in {parsed.wear} (float {low}-{high})"

    empty_hint = "Try a wider price range, another wear, or include_auctions=True."
    if parsed.wear:
        empty_hint = f"{WEAR_CAP_HINT} {empty_hint}"

    return await _run_listing_search(
        params=params,
        subject=subject,
        notes=notes,
        include_auctions=include_auctions,
        empty_hint=empty_hint,
    )


@mcp.tool()
async def ask_clarifying_questions(query: str) -> str:
    """Ask clarifying questions to help users refine generic or vague queries about CS2 skins.

    Args:
        query: The user's generic query or request

    Returns:
        Clarifying questions to help narrow down the search
    """
    query_lower = query.lower()

    # Detect various types of generic queries
    if any(word in query_lower for word in ["best", "good", "cheap", "expensive", "popular"]):
        return """To help you find the best skins, I need more details:

1. **What's your budget range?** (e.g., under $50, $100-500, over $1000)
2. **Which weapon type?** (AK-47, AWP, M4A4, Knife, Gloves, etc.)
3. **Preferred wear condition?** (Factory New, Minimal Wear, Field-Tested, Well-Worn, Battle-Scarred)
4. **Any specific skin series or theme?** (e.g., Asiimov, Redline, Fade, Case Hardened)
5. **Do you prefer StatTrak™ versions?**

Example: "Show me AK-47 skins under $100 in Factory New condition" """

    elif any(word in query_lower for word in ["knife", "knives"]):
        return """For knife searches, please specify:

1. **Knife type?** (Karambit, M9 Bayonet, Butterfly, Flip, Gut, Huntsman, etc.)
2. **Skin pattern?** (Fade, Doppler, Case Hardened, Tiger Tooth, etc.)
3. **Budget range?** (Knives vary widely from $50 to $10,000+)
4. **Wear condition preference?**
5. **StatTrak™ or regular?**

Example: "Show me Karambit Fade Factory New under $2000" """

    elif any(word in query_lower for word in ["gloves"]):
        return """For glove searches, please specify:

1. **Glove type?** (Driver Gloves, Hand Wraps, Moto Gloves, Specialist Gloves, etc.)
2. **Skin pattern?** (Crimson Weave, Fade, Case Hardened, etc.)
3. **Budget range?** (Gloves range from $50 to $5000+)
4. **Wear condition preference?**

Example: "Show me Driver Gloves Crimson Weave Minimal Wear under $500" """

    elif any(word in query_lower for word in ["investment", "profit", "trade"]):
        return """For investment/trading searches, I need to know:

1. **Your investment budget?**
2. **Risk preference?** (Stable blue skins vs volatile rare patterns)
3. **Time horizon?** (Short-term flip vs long-term hold)
4. **Skin categories of interest?** (Knives, rifles, rare stickers, etc.)

Note: I can help find current market listings but cannot provide investment advice."""

    elif len(query.split()) <= 3:
        return """Your query seems quite general. To provide better results, please specify:

1. **Weapon type** (AK-47, AWP, M4A4, Knife, etc.)
2. **Skin name or pattern** (Redline, Asiimov, Fade, etc.)
3. **Wear condition** (Factory New, Minimal Wear, etc.)
4. **Budget range** (if you have one)

Examples of specific queries:
• "AK-47 Redline Factory New under $50"
• "AWP Dragon Lore any condition"
• "Karambit Fade Minimal Wear"
• "Best AK skins under $100" """

    else:
        return f"""I understand you're looking for: "{query}"

To provide the most relevant results, could you clarify:

1. **Specific weapon or item type?**
2. **Skin name or pattern preference?**
3. **Wear condition?** (Factory New, Minimal Wear, Field-Tested, Well-Worn, Battle-Scarred)
4. **Price range?**
5. **Any other preferences?** (StatTrak™, stickers, float range, etc.)

This will help me search the CSFloat market more effectively for you!"""


@mcp.tool()
async def get_wear_conditions_help() -> str:
    """Get information about CS2 skin wear conditions and proper formatting.

    Returns:
        Information about wear conditions and skin name formatting
    """
    abbreviations = {canonical: alias for alias, canonical in WEAR_ALIASES.items() if len(alias) == 2}
    ranges = "\n".join(
        f"• {name} ({abbreviations[name].upper()}): "
        f"{WEAR_RANGES[name][0]:.2f} - {WEAR_RANGES[name][1]:.2f}"
        for name in get_wear_conditions()
    )
    return (
        "CS2 wear conditions (lower bound inclusive, upper bound exclusive):\n"
        f"{ranges}\n"
        "The abbreviations above are accepted by these tools, in any case.\n\n"
        f"{NAME_FORMAT_HINT}\n\n"
        f"{WEAR_CAP_HINT}"
    )


def main() -> None:
    """Entry point. Logging goes to stderr; stdout belongs to the JSON-RPC transport."""
    # Own handler on our own logger: the MCP SDK reconfigures root logging, and
    # this must land on stderr no matter what it does.
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(logging.Formatter("%(levelname)s %(name)s: %(message)s"))
    log.addHandler(handler)
    log.propagate = False
    log.setLevel(os.getenv("CSFLOAT_MCP_LOG_LEVEL", "INFO").upper())
    if not _api_key():
        log.error("CSFLOAT_API_KEY is not set; every tool call will return a config error.")
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()