inflation

Instructions

Implementation Reference

• src/mcp_cbr_rates/server.py:229-245 (registration)

  The MCP tool 'inflation' is registered via the @mcp.tool decorator with name='inflation'. It delegates to the _inflation (i.e., tools.inflation) function.

  @mcp.tool(
      name="inflation",
      description=(
          "Get monthly year-over-year consumer price index (CPI) inflation as"
          " published by CBR for the given year range (defaults to the previous"
          " and current year)."
      ),
  )
  async def tool_inflation(
      ctx: Context,
      year_from: int | None = None,
      year_to: int | None = None,
  ) -> InflationData:
      try:
          return await _inflation(_ctx(ctx), year_from=year_from, year_to=year_to)
      except CbrError as exc:
          raise RuntimeError(_format_error(exc)) from exc

• src/mcp_cbr_rates/tools.py:204-251 (handler)

  Main handler for the inflation tool. Fetches monthly YoY CPI inflation data from CbrClient, caches the full dataset, filters by year range, and returns an InflationData model.

  async def inflation(
      ctx: ToolContext,
      year_from: int | None = None,
      year_to: int | None = None,
  ) -> InflationData:
      """Fetch monthly year-over-year CPI inflation for the given year range."""
      today = date.today()
      if year_from is None:
          year_from = today.year - 1
      if year_to is None:
          year_to = today.year
      if year_from > year_to:
          raise CbrValidationError("year_from must be <= year_to")
  
      cache_key = "inflation:full"
      cached_all = await ctx.history_cache.get(cache_key)
      if cached_all is None:
          raw = await ctx.client.fetch_inflation()
          cached_all = raw
          await ctx.history_cache.set(cache_key, raw, ttl=DEFAULT_HISTORY_TTL)
  
      points: list[InflationPoint] = []
      for entry in cached_all:
          try:
              year = int(entry["year"])
              month = int(entry["month"])
              yoy_text = (entry.get("cpi_yoy_pct") or "").replace(",", ".").strip()
              yoy = Decimal(yoy_text) if yoy_text else None
          except Exception:
              continue
          if year < year_from or year > year_to:
              continue
          points.append(
              InflationPoint(year=year, month=month, cpi_yoy_pct=yoy, cpi_mom_pct=None)
          )
  
      points.sort(key=lambda p: (p.year, p.month))
      note = (
          "Source: cbr.ru/hd_base/infl/. CPI is measured as year-over-year change."
          if points
          else None
      )
      return InflationData(
          year_from=year_from,
          year_to=year_to,
          points=points,
          note=note,
      )

• src/mcp_cbr_rates/schemas.py:76-101 (schema)

  InflationPoint and InflationData Pydantic models — input/output schemas for the inflation tool.

  class InflationPoint(CbrModel):
      """Monthly inflation observation."""
  
      year: int = Field(..., ge=1900, le=2100)
      month: int = Field(..., ge=1, le=12)
      cpi_mom_pct: Decimal | None = Field(
          default=None,
          description="Month-over-month CPI change in percent (e.g. 0.50).",
      )
      cpi_yoy_pct: Decimal | None = Field(
          default=None,
          description="Year-over-year CPI change in percent (e.g. 7.50).",
      )
  
  
  class InflationData(CbrModel):
      """A range of monthly inflation observations."""
  
      year_from: int
      year_to: int
      points: list[InflationPoint]
      source: str = DEFAULT_SOURCE
      note: str | None = Field(
          default=None,
          description="Optional caveat (for example, when data is bundled rather than live-fetched).",
      )

• src/mcp_cbr_rates/client.py:369-412 (helper)

  CbrClient.fetch_inflation_html() and fetch_inflation() — fetches and parses the CBR inflation page using the _InflationTableParser helper.

  async def fetch_inflation_html(self) -> bytes:
      """Return the raw HTML of the public CBR inflation summary page."""
      return await self._get(INFLATION_URL)
  
  async def fetch_inflation(
      self,
  ) -> list[dict[str, str]]:
      """Return monthly inflation observations parsed from the public page.
  
      The CBR page lists year-over-year inflation in column 'Инфляция, %' for
      each month back to 2017. We do not attempt to filter — all the rows are
      returned so the caller can subset by year range.
      """
      html = await self.fetch_inflation_html()
      parser = _InflationTableParser()
      try:
          parser.feed(html.decode("utf-8", errors="replace"))
      except Exception as exc:  # pragma: no cover - parsing edge cases
          raise CbrParseError(f"failed to parse inflation HTML: {exc}") from exc
  
      results: list[dict[str, str]] = []
      for row in parser.rows:
          cells = [c for c in row if c]
          if len(cells) < 3:
              continue
          month_year = _parse_ru_month_year(cells[0])
          if month_year is None and len(cells) >= 4:
              month_year = _parse_ru_month_year(cells[1])
              value_idx = 3
          else:
              value_idx = 2
          if month_year is None:
              continue
          year, month = month_year
          inflation_text = cells[value_idx] if value_idx < len(cells) else ""
          inflation_text = re.sub(r"[^\d.,\-]", "", inflation_text)
          results.append(
              {
                  "year": str(year),
                  "month": str(month),
                  "cpi_yoy_pct": inflation_text,
              }
          )
      return results

• src/mcp_cbr_rates/client.py:116-181 (helper)

  _InflationTableParser — HTMLParser subclass that extracts rows from the CBR inflation table, and _parse_ru_month_year helper to parse Russian month names.

  class _InflationTableParser(HTMLParser):
      """Tolerant parser of the single ``<table class="data">`` on the CBR page.
  
      The expected row shape is:
      ``[empty | "Дата" | "Ключевая ставка..." | "Инфляция..." | "Цель..."]``
      (occasionally with extra leading empty cells).
  
      We accept any row with at least three textual cells where the first cell
      looks like a Russian "month YYYY" string.
      """
  
      def __init__(self) -> None:
          super().__init__()
          self._in_table = False
          self._in_row = False
          self._in_cell = False
          self._current_row: list[str] = []
          self._current_cell: list[str] = []
          self.rows: list[list[str]] = []
  
      def handle_starttag(
          self, tag: str, attrs: list[tuple[str, str | None]]
      ) -> None:
          if tag == "table":
              classes = dict(attrs).get("class") or ""
              if "data" in classes.split():
                  self._in_table = True
          elif self._in_table and tag == "tr":
              self._in_row = True
              self._current_row = []
          elif self._in_row and tag in {"td", "th"}:
              self._in_cell = True
              self._current_cell = []
  
      def handle_endtag(self, tag: str) -> None:
          if tag == "table" and self._in_table:
              self._in_table = False
          elif tag == "tr" and self._in_row:
              self._in_row = False
              if self._current_row:
                  self.rows.append(self._current_row)
          elif tag in {"td", "th"} and self._in_cell:
              self._in_cell = False
              self._current_row.append("".join(self._current_cell).strip())
  
      def handle_data(self, data: str) -> None:
          if self._in_cell:
              self._current_cell.append(data)
  
  
  def _parse_ru_month_year(raw: str) -> tuple[int, int] | None:
      """Parse strings like 'март 2026' / 'марта 2026'. Returns (year, month)."""
      if not raw:
          return None
      parts = raw.lower().replace("\xa0", " ").split()
      if len(parts) != 2:
          return None
      month_name, year_str = parts
      month = _RU_MONTHS.get(month_name)
      if not month:
          return None
      try:
          year = int(year_str)
      except ValueError:
          return None
      return year, month