Get Investment Holdings

Instructions

Implementation Reference

• server.py:337-365 (handler)

  Main handler for get_investment_holdings. Iterates over all linked items, calls Plaid's investments_holdings_get API, enriches each holding with security metadata via shape_holding, and adds institution name.

  def _get_investment_holdings_impl() -> dict:
      """Return investment holdings with security metadata across all linked Items.
  
      Joins holdings with the securities list returned in the same response to
      provide symbol, name, and security type. Adds institution to each holding.
  
      Returns:
          {"holdings": [...], "warnings": [...]}
      """
      api = build_api()
      holdings: list[dict] = []
      warnings: list[dict] = []
      for env_key, token, health in all_items(api):
          if health.status != "healthy":
              warnings.append(_warning_from_health(health))
              continue
          try:
              resp = api.investments_holdings_get(
                  InvestmentsHoldingsGetRequest(access_token=token.reveal())
              ).to_dict()
              secs_by_id = {s["security_id"]: s for s in resp.get("securities", []) or []}
              for h in resp.get("holdings", []) or []:
                  shaped = shape_holding(h, secs_by_id)
                  shaped["institution"] = health.institution_name
                  holdings.append(shaped)
          except ApiException as e:
              mapped = map_plaid_error(e, health.institution_name)["error"]
              warnings.append({"institution": health.institution_name, **mapped})
      return {"holdings": holdings, "warnings": warnings}

• server.py:368-371 (registration)

  Tool registration using FastMCP's mcp.tool decorator with readOnlyHint annotation and name 'get_investment_holdings', wrapping _get_investment_holdings_impl.

  get_investment_holdings = mcp.tool(
      annotations={"readOnlyHint": True, "title": "Get Investment Holdings"},
      name="get_investment_holdings",
  )(_get_investment_holdings_impl)

• plaid_client.py:253-271 (helper)

  Helper function shape_holding that joins a raw Plaid holding dict with security metadata (symbol, name, type) and renames fields (e.g., institution_value -> market_value).

  def shape_holding(raw_holding: dict, securities: dict[str, dict]) -> dict:
      """Shape a raw Plaid holding dict, joining with security data.
  
      - Looks up security metadata (symbol, name, type) by security_id.
      - Flattens institution_value -> market_value, institution_price -> price.
      - Handles missing security gracefully (returns None for symbol/name/type).
      """
      sec = securities.get(raw_holding.get("security_id"), {})
      return {
          "account_id": raw_holding.get("account_id"),
          "symbol": sec.get("ticker_symbol"),
          "name": sec.get("name"),
          "type": sec.get("type"),
          "quantity": raw_holding.get("quantity"),
          "cost_basis": raw_holding.get("cost_basis"),
          "market_value": raw_holding.get("institution_value"),
          "price": raw_holding.get("institution_price"),
          "currency": raw_holding.get("iso_currency_code"),
      }

• server.py:11-11 (schema)

  Imports InvestmentsHoldingsGetRequest from Plaid SDK, used by the handler to build the API request.

  from plaid.model.investments_holdings_get_request import InvestmentsHoldingsGetRequest

• server.py:19-27 (schema)

  Imports shape_holding helper and other utilities from plaid_client module used by the handler.

  from plaid_client import (
      ItemHealth,
      all_items,
      build_api,
      map_plaid_error,
      shape_account,
      shape_holding,
      shape_transaction,
  )