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

python/mcp/main_client.py
141 строка5 KB
Shaliko Salimov
documents for RAG and supported polyclinics + dou
10 дек 2025, 20:51
10 дек 2025, 20:517cd073f
from __future__ import annotations

from datetime import date, datetime

import httpx

from afisha_client import AfishaClient
from beautiful_places_client import BeautifulPlacesClient
from dou_client import DouClient
from geo_client import GeoClient
from mfc_client import MfcClient
from mypets_client import MyPetsClient
from polyclinics_client import PolyclinicsClient
from sport_events_client import SportEventsClient
from sportgrounds_client import SportgroundsClient
from yazzh import (
    AfishaEvent,
    BeautifulPlace,
    MfcOffice,
    PetShelter,
    PetVetClinic,
    PetWalkingArea,
    SportEvent,
    Sportground,
    Dou,
    Building, Polyclinic
)


class YazzhMainClient:
    """Фасад над клиентами «Я здесь живу» для MCP-методов."""

    def __init__(
            self,
            http_client: httpx.AsyncClient,
            *,
            base_url: str,
            base_geo_url: str,
            region: str = "78",
            backend_version: str | None = None,
            user_id: str | None = None,
    ) -> None:
        base_kwargs = {
            "http_client": http_client,
            "base_url": base_url,
            "base_geo_url": base_geo_url,
            "region": region,
            "backend_version": backend_version,
            "user_id": user_id,
        }
        self._afisha_client = AfishaClient(**base_kwargs)
        self._beautiful_places_client = BeautifulPlacesClient(**base_kwargs)
        self._sport_events_client = SportEventsClient(**base_kwargs)
        self._sportgrounds_client = SportgroundsClient(**base_kwargs)
        self._mypets_client = MyPetsClient(**base_kwargs)
        self._mfc_client = MfcClient(**base_kwargs)
        self._geo_client = GeoClient(**base_kwargs)
        self._dou_client = DouClient(**base_kwargs)
        self._polyclinics_client = PolyclinicsClient(**base_kwargs)

    async def search_afisha_events(
            self,
            *,
            start_date: datetime | date | str | None,
            end_date: datetime | date | str | None,
            free: bool | None = None,
            kids: bool | None = None,
            categoria: str | None = None,
    ) -> list[AfishaEvent]:
        return await self._afisha_client.list_events(
            start_date=start_date,
            end_date=end_date,
            free=free,
            kids=kids,
            categoria=categoria,
        )

    async def search_beautiful_places(
            self,
            *,
            district: str | None = None,
            categoria: str | None = None,
            keyword: str | None = None,
    ) -> list[BeautifulPlace]:
        return await self._beautiful_places_client.list_places(
            district=district,
            categoria=categoria,
            keyword=keyword,
        )

    async def search_sport_events(
            self,
            *,
            start_date: datetime | date | str | None = None,
            end_date: datetime | date | str | None = None,
            district: str | None = None,
            categoria: str | None = None,
    ) -> list[SportEvent]:
        return await self._sport_events_client.list_events(
            start_date=start_date,
            end_date=end_date,
            district=district,
            categoria=categoria,
        )

    async def search_sportgrounds(
            self,
            *,
            district: str | None = None,
            season: str | None = None,
            light: str | None = None,
    ) -> list[Sportground]:
        return await self._sportgrounds_client.list_grounds(
            district=district,
            season=season,
            light=light,
        )

    async def search_vet_clinics(self, *, district: str | None = None) -> list[PetVetClinic]:
        return await self._mypets_client.list_vet_clinics(district=district)

    async def search_pet_shelters(self, *, district: str | None = None) -> list[PetShelter]:
        return await self._mypets_client.list_shelters(district=district)

    async def search_walking_areas(self, *, district: str | None = None) -> list[PetWalkingArea]:
        return await self._mypets_client.list_walking_areas(district=district)

    async def search_mfc(self, *, district: str) -> list[MfcOffice]:
        return await self._mfc_client.list_by_district(district)

    async def search_dou(self, *, district: str, age_year: int) -> list[Dou]:
        return await self._dou_client.list_doi(district=district, age_year=age_year)

    async def search_building_info_by_id(self, building_id: str) -> Building | None:
        return await self._geo_client.get_building_info_by_building_id(building_id=building_id)

    async def search_building_info_by_address(self, address: str) -> Building | None:
        return await self._geo_client.get_building_info_by_address(address=address)

    async def search_polyclinic(self, building_id: int) -> list[Polyclinic]:
        return await self._polyclinics_client.list_polyclinics(building_id=building_id)