key_rate

Instructions

Implementation Reference

• src/mcp_cbr_rates/server.py:211-227 (registration)

  Registration of the 'key_rate' MCP tool via @mcp.tool decorator. Maps the tool to tool_key_rate async handler.

  @mcp.tool(
      name="key_rate",
      description=(
          "Get the CBR key-rate (ставка рефинансирования) time series for the"
          " requested range. Defaults to the most recent 30 days."
      ),
  )
  async def tool_key_rate(
      ctx: Context,
      date_from: date | None = None,
      date_to: date | None = None,
  ) -> KeyRateHistory:
      try:
          return await _key_rate(_ctx(ctx), date_from=date_from, date_to=date_to)
      except CbrError as exc:
          raise RuntimeError(_format_error(exc)) from exc

• src/mcp_cbr_rates/tools.py:168-201 (handler)

  Core handler for the 'key_rate' tool. Fetches CBR key-rate series for a date range (default: last 30 days), with caching and validation.

  async def key_rate(
      ctx: ToolContext,
      date_from: date | str | None = None,
      date_to: date | str | None = None,
  ) -> KeyRateHistory:
      """Fetch the CBR key-rate series for the given range (default: last 30 days)."""
      dt = _coerce_date(date_to) or date.today()
      df = _coerce_date(date_from) or (dt - timedelta(days=30))
      if df > dt:
          raise CbrValidationError("date_from must be on or before date_to")
      if (dt - df).days > 365 * 5:
          raise CbrValidationError("key_rate range cannot exceed 5 years")
  
      cache_key = f"keyrate:{df.isoformat()}:{dt.isoformat()}"
      cached = await ctx.history_cache.get(cache_key)
      if cached is not None:
          return cached  # type: ignore[no-any-return]
  
      raw = await ctx.client.fetch_key_rate(df, dt)
      points: list[KeyRatePoint] = []
      for entry in raw:
          try:
              iso = entry["date"][:10]
              point_date = date.fromisoformat(iso)
              rate = _decimal(entry["rate"])
          except Exception:
              logger.debug("skipping malformed key-rate record: %s", entry)
              continue
          points.append(KeyRatePoint(date=point_date, rate=rate))
  
      points.sort(key=lambda p: p.date)
      result = KeyRateHistory(date_from=df, date_to=dt, points=points)
      await ctx.history_cache.set(cache_key, result, ttl=DEFAULT_HISTORY_TTL)
      return result

• src/mcp_cbr_rates/schemas.py:60-65 (schema)

  KeyRatePoint schema — one observation in the CBR key-rate series.

  class KeyRatePoint(CbrModel):
      """One observation in the CBR key-rate series."""
  
      date: _dt.date
      rate: Decimal = Field(..., description="Key rate as a percentage, e.g. 16.00.")

• src/mcp_cbr_rates/schemas.py:67-73 (schema)

  KeyRateHistory response schema — key-rate observations between date_from and date_to.

  class KeyRateHistory(CbrModel):
      """Key-rate observations between ``date_from`` and ``date_to`` (inclusive)."""
  
      date_from: _dt.date
      date_to: _dt.date
      points: list[KeyRatePoint]
      source: str = DEFAULT_SOURCE

• src/mcp_cbr_rates/client.py:329-367 (helper)

  CbrClient.fetch_key_rate — makes a SOAP POST to the CBR DailyInfo service and parses <KR> elements with DT/Rate children.

  async def fetch_key_rate(
      self, date_from: date, date_to: date
  ) -> list[dict[str, str]]:
      """Return key-rate observations between ``date_from`` and ``date_to``."""
      envelope = (
          '<?xml version="1.0" encoding="utf-8"?>'
          '<soap:Envelope xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"'
          ' xmlns:xsd="http://www.w3.org/2001/XMLSchema"'
          ' xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">'
          "<soap:Body>"
          '<KeyRateXML xmlns="http://web.cbr.ru/">'
          f"<fromDate>{date_from.isoformat()}</fromDate>"
          f"<ToDate>{date_to.isoformat()}</ToDate>"
          "</KeyRateXML>"
          "</soap:Body>"
          "</soap:Envelope>"
      ).encode()
      headers = {
          "Content-Type": "text/xml; charset=utf-8",
          "SOAPAction": "http://web.cbr.ru/KeyRateXML",
      }
      payload = await self._post(SOAP_URL, envelope, headers=headers)
      root = _parse_xml_bytes(payload)
      # Strip the SOAP envelope to find <KR> nodes anywhere underneath.
      results: list[dict[str, str]] = []
      for kr in root.iter():
          if kr.tag.split("}")[-1] != "KR":
              continue
          dt = ""
          rate = ""
          for child in kr:
              tag = child.tag.split("}")[-1]
              if tag == "DT":
                  dt = (child.text or "").strip()
              elif tag == "Rate":
                  rate = (child.text or "").strip()
          if dt and rate:
              results.append({"date": dt, "rate": rate})
      return results