Get Transactions

Instructions

Implementation Reference

• server.py:132-182 (handler)

  The main implementation function _get_transactions_impl that fetches transactions across all healthy Plaid items using offset pagination (count=500 per page), applies date window clipping, optionally filters by account_ids, and handles API errors via warnings.

  def _get_transactions_impl(
      start_date: str,
      end_date: str,
      account_ids: list[str] | None = None,
  ) -> dict:
      """Fetch transactions in [start_date, end_date] across all healthy Items.
  
      Dates are ISO YYYY-MM-DD. Uses Plaid /transactions/get with 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.
      """
      api = build_api()
      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})
  
      base_options: dict = {"count": 500}
      if account_ids:
          base_options["account_ids"] = list(account_ids)
  
      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:
                  options = TransactionsGetRequestOptions(**{**base_options, "offset": offset})
                  resp = api.transactions_get(
                      TransactionsGetRequest(
                          access_token=token.reveal(),
                          start_date=date.fromisoformat(clipped_start),
                          end_date=date.fromisoformat(clipped_end),
                          options=options,
                      )
                  ).to_dict()
                  batch = resp.get("transactions", []) or []
                  for raw in batch:
                      transactions.append(shape_transaction(raw))
                  total = resp.get("total_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 {"transactions": transactions, "warnings": warnings}

• server.py:185-188 (registration)

  Registration of 'get_transactions' as an MCP tool using mcp.tool(), decorating _get_transactions_impl with annotations including readOnlyHint and title.

  get_transactions = mcp.tool(
      annotations={"readOnlyHint": True, "title": "Get Transactions"},
      name="get_transactions",
  )(_get_transactions_impl)

• server.py:14-16 (schema)

  Imports of TransactionsGetRequest and TransactionsGetRequestOptions from Plaid model, which define the request schema for the Plaid API call inside the tool.

  from plaid.model.liabilities_get_request import LiabilitiesGetRequest
  from plaid.model.transactions_get_request import TransactionsGetRequest
  from plaid.model.transactions_get_request_options import TransactionsGetRequestOptions

• plaid_client.py:232-250 (helper)

  The shape_transaction helper function that normalizes raw Plaid transaction data into a trimmed response, preferring personal_finance_category over legacy category and falling back merchant_name to name.

  def shape_transaction(raw: dict) -> dict:
      """Shape a raw Plaid transaction dict into a trimmed, normalized response.
  
      - Prefers personal_finance_category (primary/detailed) over legacy category.
      - Extracts merchant_name or falls back to name.
      - Includes pending status and currency.
      """
      pfc = raw.get("personal_finance_category") or {}
      return {
          "transaction_id": raw.get("transaction_id"),
          "account_id": raw.get("account_id"),
          "date": str(raw.get("date")) if raw.get("date") else None,
          "amount": raw.get("amount"),
          "currency": raw.get("iso_currency_code"),
          "merchant": raw.get("merchant_name") or raw.get("name"),
          "name": raw.get("name"),
          "category": {"primary": pfc.get("primary"), "detailed": pfc.get("detailed")},
          "pending": bool(raw.get("pending")),
      }

• server.py:116-129 (helper)

  The _clip_window helper that enforces a ~2-year max lookback window, returning clipped start/end dates and an optional warning reason.

  _MAX_LOOKBACK_DAYS = 730  # ~2 years
  
  
  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