Get Recurring Transactions

Instructions

Implementation Reference

• server.py:208-244 (handler)

  Core implementation of get_recurring_transactions. Calls Plaid's /accounts/get per Item to collect account IDs, then fetches recurring streams via /transactions/recurring/get. Returns unified inflows/outflows lists with warnings for unhealthy Items or API errors.

  def _get_recurring_transactions_impl() -> dict:
      """Return recurring inflow and outflow streams across all linked Items.
  
      Calls /accounts/get first per Item to collect account IDs (required by
      /transactions/recurring/get), then fetches recurring streams and shapes
      them into unified inflows/outflows lists.
  
      Returns:
          {"inflows": [...], "outflows": [...], "warnings": [...]}
      """
      api = build_api()
      inflows: list[dict] = []
      outflows: 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:
              acct_resp = api.accounts_get(
                  AccountsGetRequest(access_token=token.reveal())
              ).to_dict()
              account_ids = [a["account_id"] for a in acct_resp.get("accounts", [])]
              resp = api.transactions_recurring_get(
                  TransactionsRecurringGetRequest(
                      access_token=token.reveal(),
                      account_ids=account_ids,
                  )
              ).to_dict()
              for stream in resp.get("inflow_streams", []) or []:
                  inflows.append(_shape_stream(stream, health.institution_name))
              for stream in resp.get("outflow_streams", []) or []:
                  outflows.append(_shape_stream(stream, health.institution_name))
          except ApiException as e:
              mapped = map_plaid_error(e, health.institution_name)["error"]
              warnings.append({"institution": health.institution_name, **mapped})
      return {"inflows": inflows, "outflows": outflows, "warnings": warnings}

• server.py:247-250 (registration)

  Registration of the tool with FastMCP. Decorates _get_recurring_transactions_impl with name='get_recurring_transactions' and read-only hint annotation.

  get_recurring_transactions = mcp.tool(
      annotations={"readOnlyHint": True, "title": "Get Recurring Transactions"},
      name="get_recurring_transactions",
  )(_get_recurring_transactions_impl)

• server.py:191-205 (helper)

  Helper function _shape_stream that transforms a raw Plaid recurring stream dict into a canonical shape (stream_id, description, merchant, average_amount, frequency, last_date, is_active, category, account_id, institution).

  def _shape_stream(stream: dict, institution: str | None) -> dict:
      avg_amount_obj = stream.get("average_amount") or {}
      pfc = stream.get("personal_finance_category") or {}
      return {
          "stream_id": stream.get("stream_id"),
          "description": stream.get("description"),
          "merchant": stream.get("merchant_name"),
          "average_amount": avg_amount_obj.get("amount"),
          "frequency": stream.get("frequency"),
          "last_date": str(stream.get("last_date")) if stream.get("last_date") else None,
          "is_active": stream.get("is_active"),
          "category": pfc.get("primary"),
          "account_id": stream.get("account_id"),
          "institution": institution,
      }

• server.py:17-17 (schema)

  Import of Plaid's TransactionsRecurringGetRequest model used in the handler.

  from plaid.model.transactions_recurring_get_request import TransactionsRecurringGetRequest