yfinance_get_financials

Instructions

Implementation Reference

• src/yfmcp/server.py:690-698 (registration)

  Tool registration via @mcp.tool() decorator with name 'yfinance_get_financials', readonly, idempotent hint.

  @mcp.tool(
      name="yfinance_get_financials",
      annotations=ToolAnnotations(
          readOnlyHint=True,
          destructiveHint=False,
          idempotentHint=True,
          openWorldHint=True,
      ),
  )

• src/yfmcp/server.py:699-773 (handler)

  Async handler 'get_financials' that fetches income statement, balance sheet, and cash flow for a given symbol/frequency.

  async def get_financials(
      symbol: Annotated[str, Field(description="Stock ticker symbol (e.g., 'AAPL', 'GOOGL', 'MSFT')")],
      frequency: Annotated[
          str,
          Field(
              description=(
                  "Reporting frequency: 'annual' for yearly, 'quarterly' for quarterly, "
                  "or 'ttm' for trailing twelve months"
              )
          ),
      ] = "annual",
  ) -> str:
      """Fetch financial statements (income statement, balance sheet, and cash flow) with historical data.
  
      Returns JSON with income statement, balance sheet, and cash flow data across reporting periods.
  
      Use the data to analyze trends, calculate ratios, or compare periods.
      """
      try:
          ticker = await asyncio.to_thread(yf.Ticker, symbol)
      except _RETRYABLE_YFINANCE_EXCEPTIONS as exc:
          return _create_retryable_error_response(f"fetching financials for '{symbol}'", exc, {"symbol": symbol})
      except Exception as exc:
          return create_error_response(
              f"Failed to fetch financials for '{symbol}'. Verify the symbol is correct.",
              error_code="API_ERROR",
              details={"symbol": symbol, "exception": str(exc)},
          )
  
      income_stmt = None
      balance_sheet = None
      cash_flow = None
  
      if frequency not in {"annual", "quarterly", "ttm"}:
          return create_error_response(
              f"Invalid frequency '{frequency}'. Valid options: 'annual', 'quarterly', 'ttm'.",
              error_code="INVALID_PARAMS",
              details={"frequency": frequency, "valid_options": ["annual", "quarterly", "ttm"]},
          )
  
      try:
          if frequency == "annual":
              income_stmt = await asyncio.to_thread(lambda: ticker.income_stmt)
              balance_sheet = await asyncio.to_thread(lambda: ticker.balance_sheet)
              cash_flow = await asyncio.to_thread(lambda: ticker.cashflow)
          elif frequency == "quarterly":
              income_stmt = await asyncio.to_thread(lambda: ticker.quarterly_income_stmt)
              balance_sheet = await asyncio.to_thread(lambda: ticker.quarterly_balance_sheet)
              cash_flow = await asyncio.to_thread(lambda: ticker.quarterly_cashflow)
          else:
              income_stmt = await asyncio.to_thread(lambda: ticker.ttm_income_stmt)
              balance_sheet = None  # TTM balance sheet not directly available
              cash_flow = None  # TTM cash flow not directly available
  
          result = _build_financials_response(income_stmt, balance_sheet, cash_flow)
      except _RETRYABLE_YFINANCE_EXCEPTIONS as exc:
          return _create_retryable_error_response(
              f"fetching financials for '{symbol}'",
              exc,
              {"symbol": symbol, "frequency": frequency},
          )
      except Exception as exc:
          return create_error_response(
              f"Failed to fetch financials for '{symbol}'. Verify the symbol is correct.",
              error_code="API_ERROR",
              details={"symbol": symbol, "frequency": frequency, "exception": str(exc)},
          )
      if not result:
          return create_error_response(
              f"No financial data available for '{symbol}' with frequency='{frequency}'.",
              error_code="NO_DATA",
              details={"symbol": symbol, "frequency": frequency},
          )
  
      return dump_json(result)

• src/yfmcp/server.py:776-833 (helper)

  Helper '_build_financials_response' transforms DataFrames into a dict with income_statement, balance_sheet, and cash_flow fields.

  def _build_financials_response(income_stmt, balance_sheet, cash_flow=None) -> dict:
      """Build financials response from income statement, balance sheet, and cash flow DataFrames."""
      result = {}
  
      if income_stmt is not None and not income_stmt.empty:
          income_fields = [
              "EBIT",
              "Net Income",
              "Tax Provision",
              "Pretax Income",
              "Interest Expense",
              "Total Revenue",
              "Operating Income",
              "EBITDA",
              "Normalized Income",
          ]
          available_income_fields = [f for f in income_fields if f in income_stmt.index]
          result["income_statement"] = {}
          for field in available_income_fields:
              result["income_statement"][field] = {
                  str(col.date()): income_stmt.loc[field, col] for col in income_stmt.columns
              }
  
      if balance_sheet is not None and not balance_sheet.empty:
          balance_fields = [
              "Stockholders Equity",
              "Total Debt",
              "Cash And Cash Equivalents",
              "Invested Capital",
              "Net Debt",
              "Total Assets",
              "Total Liabilities Net Minority Interest",
              "Net Tangible Assets",
              "Tangible Book Value",
          ]
          available_balance_fields = [f for f in balance_fields if f in balance_sheet.index]
          result["balance_sheet"] = {}
          for field in available_balance_fields:
              result["balance_sheet"][field] = {
                  str(col.date()): balance_sheet.loc[field, col] for col in balance_sheet.columns
              }
  
      if cash_flow is not None and not cash_flow.empty:
          cash_flow_fields = [
              "Operating Cash Flow",
              "Free Cash Flow",
              "Capital Expenditure",
              "Net Income From Continuing Operations",
              "Depreciation And Amortization",
              "Change In Working Capital",
              "Cash Dividends Paid",
          ]
          available_cash_flow_fields = [f for f in cash_flow_fields if f in cash_flow.index]
          result["cash_flow"] = {}
          for field in available_cash_flow_fields:
              result["cash_flow"][field] = {str(col.date()): cash_flow.loc[field, col] for col in cash_flow.columns}
  
      return result

• src/yfmcp/server.py:46-52 (helper)

  Helper '_create_retryable_error_response' creates standardized retryable/rate-limit error responses.

  def _create_retryable_error_response(action: str, exc: BaseException, details: dict[str, Any]) -> str:
      if _is_rate_limit_error(exc):
          message = f"Rate limit reached while {action}. Try again later."
      else:
          message = f"Temporary network issue while {action}. Try again later."
  
      return create_error_response(message, error_code="NETWORK_ERROR", details={**details, "exception": str(exc)})

• src/yfmcp/utils.py:10-24 (schema)

  Helper 'create_error_response' and 'dump_json' used for error formatting and JSON serialization.

  def create_error_response(message: str, error_code: ErrorCode = "UNKNOWN_ERROR", details: dict | None = None) -> str:
      """Create a structured error response.
  
      Args:
          message: Human-readable error message
          error_code: Machine-readable error code for client handling
          details: Optional additional error details
  
      Returns:
          JSON string with error information
      """
      error_obj: dict[str, object] = {"error": message, "error_code": error_code}
      if details:
          error_obj["details"] = details
      return dump_json(error_obj)