python/mcp/mcp_server.py - master - ksilisk/spbtechrun_hack

python/mcp/mcp_server.py
268 строк11 KB
Shaliko Salimov
user profile support
12 дек 2025, 21:13
12 дек 2025, 21:1316730d4
from datetime import datetime, timedelta, date
from typing import List, Annotated, Optional

from mcp.server.fastmcp import FastMCP
from pydantic import Field

from config import (
    MCP_HOST,
    MCP_PORT,
    YAZZH_BACKEND_VERSION,
    YAZZH_BASE_URL,
    YAZZH_REGION,
    YAZZH_USER_ID,
    YAZZH_GEO_BASE_URL
)
from http_client import init_http_client
from main_client import YazzhMainClient
from yazzh import (
    AfishaEvent,
    BeautifulPlace,
    MfcOffice,
    PetShelter,
    PetVetClinic,
    PetWalkingArea,
    SportEvent,
    Sportground, DouFillInfo, Polyclinic,
)

# Рекомендуемая конфигурация для HTTP:
# stateless_http=True + json_response=True
mcp = FastMCP(
    "city-assistant",
    stateless_http=True,
    # json_response=True,
    host=MCP_HOST,
    port=MCP_PORT,
    debug=True,
)

_main_client: Optional[YazzhMainClient] = None


async def get_main_client() -> YazzhMainClient:
    global _main_client
    if _main_client is None:
        http_client = await init_http_client()
        _main_client = YazzhMainClient(
            http_client,
            base_url=YAZZH_BASE_URL,
            base_geo_url=YAZZH_GEO_BASE_URL,
            region=YAZZH_REGION,
            backend_version=YAZZH_BACKEND_VERSION,
            user_id=YAZZH_USER_ID,
        )
    return _main_client


def _to_bool(value: str) -> bool | None:
    if not value:
        return None
    lowered = value.lower()
    if lowered in {"true", "1", "yes", "y"}:
        return True
    if lowered in {"false", "0", "no", "n"}:
        return False
    return None


@mcp.tool(
    name="search_afisha_events",
    description="Поиск афиши по заданным параметрам от пользователя",
)
async def search_afisha_events(
        start_date: Annotated[str, Field(description="Дата начала, формат 'YYYY-MM-DD'")] = "",
        end_date: Annotated[str, Field(description="Дата конца, формат 'YYYY-MM-DD'")] = "",
        free: Annotated[
            str, Field(description="True — только бесплатные. False - только платные. None — без фильтра.")] = None,
        kids: Annotated[str, Field(
            description="Фильтр по детским мероприятиям. True - только для детей, False - не для детей, None - не задано")] = None
) -> List[AfishaEvent]:
    """
    Дергаем /afisha/all с параметрами.
    """
    print(
        f"Invoked 'search_afisha_events' with start_date='{start_date}', end_date='{end_date}', free='{free}', kids='{kids}'")
    if start_date == "":
        start_date = str(datetime.today().isoformat(timespec="seconds"))
    if end_date == "":
        end_date = str((datetime.today() + timedelta(days=7)).isoformat(timespec="seconds"))
    client = await get_main_client()
    return await client.search_afisha_events(
        start_date=start_date or None,
        end_date=end_date or None,
        free=_to_bool(free),
        kids=_to_bool(kids) or None,
    )


@mcp.tool(
    name="search_sportgrounds",
    description="Поиск спортивных площадок по заданным параметрам от пользователя",
)
async def search_sportgrounds(
        district: Annotated[str, Field(description="Район, в котором надо искать площадки")] = None,
        season: Annotated[str, Field(
            description="Сезон активности - площадки для какого времени года искать (ЗИМА, ЛЕТО, ВСЕ)")] = None,
        light: Annotated[str, Field(
            description="Наличие освещения на площадке. True - есть освещение, False - нет освещения, пустая строка - не задано")] = None
) -> List[Sportground]:
    """
    Дергаем /sportgrounds с параметрами.
    """
    print(f"Invoked 'search_sportgrounds' with district='{district}', season='{season}', light='{light}'")
    client = await get_main_client()
    return await client.search_sportgrounds(
        district=district or None,
        season=season or None,
        light=light or None,
    )


@mcp.tool(
    name="search_sport_events",
    description="Поиск спортивных событий по заданным параметрам от пользователя",
)
async def search_sport_events(
        start_date: Annotated[str, Field(description="Дата начала, формат 'YYYY-MM-DD'")] = "",
        end_date: Annotated[str, Field(description="Дата конца, формат 'YYYY-MM-DD'")] = "",
        district: Annotated[str, Field(description="Район, в котором надо спортивные события")] = "",
        categoria: Annotated[str, Field(description="Вид спорта, для которого надо искать события")] = ""
) -> List[SportEvent]:
    """
    Дергаем /sport-events с параметрами.
    """
    print(
        f"Invoked 'search_sport_events' with start_date='{start_date}', end_date='{end_date}', district='{district}', categoria='{categoria}'")
    if start_date == "":
        start_date = str(date.today().isoformat())
    if end_date == "":
        end_date = str((date.today() + timedelta(days=15)).isoformat())
    client = await get_main_client()
    return await client.search_sport_events(
        start_date=start_date or None,
        end_date=end_date or None,
        district=district or None,
        categoria=categoria or None,
    )


@mcp.tool(
    name="search_beautiful_places",
    description="Поиск красивых мест по заданным параметрам от пользователя",
)
async def search_beautiful_places(
        district: Annotated[str, Field(description="Район, в котором надо искать красивые места")] = "",
        categoria: Annotated[str, Field(description="Какой вид красивого места. Возможные значения: "
                                                    "[Природа | Архитектура | Развлечения | Гастрономия | Пустая строка]")] = "",
        keyword: Annotated[str, Field(description="Ключевое слово, по которому искать красивое место. "
                                                  "Возможные значения: [озеро | сад | архитектура | скала | пустая строка]")] = ""
) -> List[BeautifulPlace]:
    """
    Дергаем /beautiful_places с параметрами.
    """
    print(
        f"Invoked 'search_beautiful_places' with keyword='{keyword}', district='{district}', categoria='{categoria}'")
    client = await get_main_client()
    return await client.search_beautiful_places(
        district=district or None,
        categoria=categoria or None,
        keyword=keyword or None,
    )


@mcp.tool(
    name="search_veterinary_clinics",
    description="Поиск поиск ветеринарных клиник (ветклиник)",
)
async def search_veterinary_clinics(
        district: Annotated[str, Field(description="Опционально, район для фильтрации")] = "",
) -> List[PetVetClinic]:
    """
    Дергаем /mypets/all-category с type = Ветклиника.
    """
    print("Invoked 'search_veterinary_clinics'")
    client = await get_main_client()
    return await client.search_vet_clinics(district=district or None)


@mcp.tool(
    name="search_animal_shelters",
    description="Поиск поиск приютов для животных",
)
async def search_animal_shelters(
        district: Annotated[str, Field(description="Опционально, район для фильтрации")] = "",
) -> List[PetShelter]:
    """
    Дергаем /mypets/all-category с type = Приют.
    """
    print("Invoked 'search_animal_shelters'")
    client = await get_main_client()
    return await client.search_pet_shelters(district=district or None)


@mcp.tool(
    name="search_animal_walking_areas",
    description="Поиск площадок для выгула животных",
)
async def search_animal_walking_areas(
        district: Annotated[str, Field(description="Опционально, район для фильтрации")] = "",
) -> List[PetWalkingArea]:
    """
    Дергаем /mypets/all-category с type = Площадка.
    """
    print("Invoked 'search_animal_walking_areas'")
    client = await get_main_client()
    return await client.search_walking_areas(district=district or None)


@mcp.tool(
    name="search_mfc_offices",
    description="Поиск МФЦ (Многофункциональных центров) в конкретном районе пользователя",
)
async def search_mfc_offices(
        district: Annotated[str, Field(description="ОБЯЗАТЕЛЬНЫЙ ПАРАМЕТР, НЕ ПУСТАЯ СТРОКА. Район пользователя")] = ""
) -> List[MfcOffice]:
    """
    Дергаем /mfc/district с параметром district.
    """
    print(f"Invoked 'search_mfc_offices' with district='{district}'")
    client = await get_main_client()
    return await client.search_mfc(district=district)


@mcp.tool(
    name="search_dou",
    description="Поиск детских садов по заданным параметрам от пользователя"
)
async def search_dou(
        district: Annotated[str, Field(description="Район пользователя")] = "",
        age_year: Annotated[int, Field(description="Возраст ребенка")] = 0
) -> List[DouFillInfo]:
    print(f"Invoked 'search_dou' with district='{district}, age_year='{age_year}'")
    client = await get_main_client()
    dous = await client.search_dou(district=district, age_year=age_year)
    dou_results = []
    for dou in dous:
        building = await client.search_building_info_by_id(building_id=dou.building_id)
        if building is not None:
            dou_results.append(DouFillInfo(
                name=dou.name,
                address=building.address,
            ))
    return dou_results


@mcp.tool(
    name="search_polyclinics",
    description="Поиск поликлиник рядом с адресом пользователя"
)
async def search_polyclinics(
        address: Annotated[str, Field(description="ОБЯЗАТЕЛЬНЫЙ ПАРАМЕТР, НЕ ПУСТАЯ СТРОКА. Адрес пользователя")] = ""
) -> List[Polyclinic]:
    print(f"Invoked 'search_polyclinics' with address='{address}'")
    client = await get_main_client()
    building_id = await client.search_building_info_by_address(address=address)
    if building_id is None:
        return []
    return await client.search_polyclinic(building_id=building_id.id)