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

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

from typing import Any, Mapping

import httpx


class YazzhApiError(RuntimeError):
    """Ошибки, возникающие при обращении к API «Я Здесь Живу»."""


class YazzhBaseClient:
    """
    Базовый клиент: хранит общие заголовки (регион), базовый URL и вспомогательный GET-хелпер.
    Остальные клиенты наследуются от него.
    """

    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,
            default_timeout: float = 10.0,
    ) -> None:
        self._http = http_client
        self._base_url = base_url.rstrip("/")
        self._base_geo_url = base_geo_url.rstrip("/")
        self._region = region
        self._backend_version = backend_version
        self._user_id = user_id
        self._timeout = default_timeout

    def _headers(self, extra: Mapping[str, str] | None = None) -> dict[str, str]:
        headers: dict[str, str] = {"region": self._region}
        if self._backend_version:
            headers["backend-version"] = self._backend_version
        if self._user_id:
            headers["user-id"] = self._user_id
        if extra:
            headers.update(extra)
        return headers

    async def _get(
            self,
            path: str,
            *,
            params: Mapping[str, Any] | None = None,
            headers: Mapping[str, str] | None = None,
    ) -> Any:
        url = path if path.startswith("http") else f"{self._base_url}/{path.lstrip('/')}"
        try:
            response = await self._http.get(
                url,
                params={k: v for k, v in (params or {}).items() if v is not None and v != ""},
                headers=self._headers(headers),
                timeout=self._timeout,
            )
            if response.status_code == httpx.codes.NO_CONTENT:
                return {}
            response.raise_for_status()
        except httpx.HTTPStatusError as exc:  # pragma: no cover - straightforward pass-through
            print(YazzhApiError(f"Yazzh API error {exc.response.status_code}: {exc.response.text}"))
            return {}
        except httpx.HTTPError as exc:  # pragma: no cover
            # include the exception class to make diagnostics clearer (e.g. ConnectTimeout)
            print(YazzhApiError(f"Yazzh transport error ({exc.__class__.__name__}): {exc}"))
            return {}
        return response.json()

    async def _get_geo(
            self,
            path: str,
            *,
            params: Mapping[str, Any] | None = None,
            headers: Mapping[str, str] | None = None,
    ) -> Any:
        url = path if path.startswith("http") else f"{self._base_geo_url}/{path.lstrip('/')}"
        try:
            response = await self._http.get(
                url,
                params={k: v for k, v in (params or {}).items() if v is not None and v != ""},
                headers=self._headers(headers),
                timeout=self._timeout,
            )
            if response.status_code == httpx.codes.NO_CONTENT:
                return {}
            response.raise_for_status()
        except httpx.HTTPStatusError as exc:  # pragma: no cover - straightforward pass-through
            print(YazzhApiError(f"Yazzh API error {exc.response.status_code}: {exc.response.text}"))
            return {}
        except httpx.HTTPError as exc:  # pragma: no cover
            # include the exception class to make diagnostics clearer (e.g. ConnectTimeout)
            print(YazzhApiError(f"Yazzh transport error ({exc.__class__.__name__}): {exc}"))
            return {}
        return response.json()