Get Investment Transactions

Instructions

Implementation Reference

• server.py:374-443 (handler)

  The main handler implementation for the get_investment_transactions tool. Fetches investment transactions from Plaid API across all healthy Items with offset pagination, clipping date windows to ~2 years, and joining with security metadata.

  def _get_investment_transactions_impl(
      start_date: str,
      end_date: str,
  ) -> dict:
      """Fetch investment transactions in [start_date, end_date] across all healthy Items.
  
      Dates are ISO YYYY-MM-DD. Uses offset pagination (count=500 per page).
      If start_date is older than ~2 years before end_date, the window is clipped
      and a warning is emitted. Each transaction is joined with security metadata
      (ticker symbol, name) from the same response.
  
      Returns:
          {"investment_transactions": [...], "warnings": [...]}
      """
      api = build_api()
      investment_transactions: list[dict] = []
      warnings: list[dict] = []
  
      clipped_start, clipped_end, clip_reason = _clip_window(start_date, end_date)
      if clip_reason:
          warnings.append({"code": "WINDOW_CLIPPED", "reason": clip_reason, "message": clip_reason})
  
      for env_key, token, health in all_items(api):
          if health.status != "healthy":
              warnings.append(_warning_from_health(health))
              continue
          offset = 0
          try:
              while True:
                  resp = api.investments_transactions_get(
                      InvestmentsTransactionsGetRequest(
                          access_token=token.reveal(),
                          start_date=date.fromisoformat(clipped_start),
                          end_date=date.fromisoformat(clipped_end),
                          options=InvestmentsTransactionsGetRequestOptions(
                              count=500,
                              offset=offset,
                          ),
                      )
                  ).to_dict()
                  secs_by_id = {
                      s["security_id"]: s
                      for s in resp.get("securities", []) or []
                  }
                  batch = resp.get("investment_transactions", []) or []
                  for t in batch:
                      investment_transactions.append({
                          "investment_transaction_id": t.get("investment_transaction_id"),
                          "account_id": t.get("account_id"),
                          "date": str(t.get("date")) if t.get("date") else None,
                          "type": t.get("type"),
                          "subtype": t.get("subtype"),
                          "amount": t.get("amount"),
                          "quantity": t.get("quantity"),
                          "price": t.get("price"),
                          "fees": t.get("fees"),
                          "currency": t.get("iso_currency_code"),
                          "symbol": secs_by_id.get(t.get("security_id"), {}).get("ticker_symbol"),
                          "name": secs_by_id.get(t.get("security_id"), {}).get("name"),
                          "institution": health.institution_name,
                      })
                  total = resp.get("total_investment_transactions") or 0
                  offset += len(batch)
                  if offset >= total or not batch:
                      break
          except ApiException as e:
              mapped = map_plaid_error(e, health.institution_name)["error"]
              warnings.append({"institution": health.institution_name, **mapped})
  
      return {"investment_transactions": investment_transactions, "warnings": warnings}

• server.py:446-449 (registration)

  Registers _get_investment_transactions_impl as the MCP tool named 'get_investment_transactions' with readOnlyHint and title annotations.

  get_investment_transactions = mcp.tool(
      annotations={"readOnlyHint": True, "title": "Get Investment Transactions"},
      name="get_investment_transactions",
  )(_get_investment_transactions_impl)

• server.py:119-129 (helper)

  Helper function that clips the date window to a maximum of ~2 years lookback (730 days), returning the clipped start date and a warning reason if clipping occurred. Used by the investment transactions handler.

  def _clip_window(start_date: str, end_date: str) -> tuple[str, str, str | None]:
      """Return (start, end, warning_reason_or_None) clipped to the 2-year window."""
      start = date.fromisoformat(start_date)
      end = date.fromisoformat(end_date)
      earliest = end - timedelta(days=_MAX_LOOKBACK_DAYS)
      if start < earliest:
          return earliest.isoformat(), end.isoformat(), (
              f"clipped start from {start.isoformat()} to {earliest.isoformat()} "
              "(Plaid max lookback ~2 years)"
          )
      return start.isoformat(), end.isoformat(), None

• server.py:32-37 (helper)

  Helper that formats a warning dict from an ItemHealth object for unhealthy items. Used by the handler when skipping unhealthy Items.

  def _warning_from_health(h: ItemHealth) -> dict:
      return {
          "institution": h.institution_name or h.env_key,
          "status": h.status,
          "reason": h.reason,
      }

• server.py:116-116 (helper)

  Constant defining the maximum lookback window of 730 days (~2 years) used for clipping date ranges in investment transactions (and other) queries.

  _MAX_LOOKBACK_DAYS = 730  # ~2 years