get_financials

Instructions

Implementation Reference

• src/massive_mcp/tools/financials.py:14-37 (handler)

  The 'get_financials' async tool handler function. Accepts ticker, timeframe (annual/quarterly/ttm), limit, and cursor; calls the Massive API endpoint /vX/reference/financials with query params and returns the response dict.

  async def get_financials(
      ticker: str,
      timeframe: Timeframe = "quarterly",
      limit: int = 4,
      cursor: str | None = None,
  ) -> dict[str, Any]:
      """Financial statements (income, balance sheet, cash flow, ratios) for a ticker.
  
      Args:
          ticker: Stock symbol.
          timeframe: "annual", "quarterly", or "ttm". Default "quarterly".
          limit: Max periods returned. Default 4 (last year).
          cursor: Pagination cursor.
      """
      return await client.get(
          "/vX/reference/financials",
          {
              "ticker": ticker,
              "timeframe": timeframe,
              "limit": limit,
              "cursor": cursor,
              "order": "desc",
          },
      )

• src/massive_mcp/tools/financials.py:9-9 (schema)

  Timeframe type alias (Literal['annual', 'quarterly', 'ttm']) used for the timeframe parameter of get_financials.

  Timeframe = Literal["annual", "quarterly", "ttm"]

• src/massive_mcp/server.py:37-48 (registration)

  Registration loop iterates over tool modules (including financials) and calls module.register(mcp, client) to register all tools with the FastMCP server.

  for module in (
      aggregates,
      quotes,
      snapshots,
      tickers,
      news,
      reference,
      indicators,
      corporate,
      financials,
  ):
      module.register(mcp, client)

• src/massive_mcp/tools/financials.py:12-37 (registration)

  The register() function that uses the @mcp.tool() decorator to register get_financials (and other tools) with FastMCP.

  def register(mcp: FastMCP, client: MassiveClient) -> None:
      @mcp.tool()
      async def get_financials(
          ticker: str,
          timeframe: Timeframe = "quarterly",
          limit: int = 4,
          cursor: str | None = None,
      ) -> dict[str, Any]:
          """Financial statements (income, balance sheet, cash flow, ratios) for a ticker.
  
          Args:
              ticker: Stock symbol.
              timeframe: "annual", "quarterly", or "ttm". Default "quarterly".
              limit: Max periods returned. Default 4 (last year).
              cursor: Pagination cursor.
          """
          return await client.get(
              "/vX/reference/financials",
              {
                  "ticker": ticker,
                  "timeframe": timeframe,
                  "limit": limit,
                  "cursor": cursor,
                  "order": "desc",
              },
          )

• src/massive_mcp/client.py:47-89 (helper)

  MassiveClient.get() - the HTTP helper that executes the actual API call; get_financials delegates to this method.

  async def get(
      self, path: str, params: dict[str, Any] | None = None, *, trim: bool = True
  ) -> dict[str, Any]:
      merged: dict[str, Any] = {k: v for k, v in (params or {}).items() if v is not None}
      if self._settings.auth_mode == "query":
          merged["apiKey"] = self._settings.api_key
  
      last_exc: Exception | None = None
      for attempt in range(MAX_RETRIES):
          try:
              resp = await self._http.get(path, params=merged)
          except httpx.HTTPError as exc:
              last_exc = exc
              await asyncio.sleep(2**attempt)
              continue
  
          if resp.status_code == 429:
              retry_after = float(resp.headers.get("Retry-After", 2**attempt))
              await asyncio.sleep(min(retry_after, 30))
              continue
          if 500 <= resp.status_code < 600 and attempt < MAX_RETRIES - 1:
              await asyncio.sleep(2**attempt)
              continue
  
          if resp.status_code == 401:
              hint = (
                  "auth rejected — verify MASSIVE_API_KEY; "
                  "if you used MASSIVE_AUTH_MODE=bearer, try 'query' (or vice versa)"
              )
              raise MassiveAPIError(401, hint, _strip_secrets(str(resp.request.url)))
  
          try:
              data = resp.json()
          except ValueError:
              data = {"raw": resp.text}
  
          if not resp.is_success:
              msg = data.get("error") or data.get("message") or resp.reason_phrase or "request failed"
              raise MassiveAPIError(resp.status_code, str(msg), _strip_secrets(str(resp.request.url)))
  
          return _trim(data) if trim else data
  
      raise MassiveAPIError(0, f"network error after {MAX_RETRIES} retries: {last_exc}", path)