Search Transactions

Instructions

Implementation Reference

• server.py:481-546 (handler)

  The _search_transactions_impl function is the core handler for the search_transactions tool. It fetches transactions via Plaid API, filters them by case-insensitive substring match against merchant_name, name, and counterparty names, clips the date window to ~2 years, and returns shaped transactions plus warnings.

  def _search_transactions_impl(
      query: str,
      start_date: str,
      end_date: str,
  ) -> dict:
      """Search transactions by keyword across merchant, name, and counterparty names.
  
      Fetches transactions in [start_date, end_date] and filters them with a
      case-insensitive substring match against:
      - ``merchant_name``
      - ``name``
      - ``counterparties[].name``
  
      The match is performed on the raw Plaid payload before shaping so that
      counterparty names (which are dropped by ``shape_transaction``) are
      searchable. Dates are ISO YYYY-MM-DD. The window is clipped to ~2 years
      and a WINDOW_CLIPPED warning is emitted when applicable.
  
      Returns:
          {"transactions": [...], "warnings": [...]}
      """
      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})
  
      q = query.lower()
  
      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(count=500, 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:
                      merchant = (raw.get("merchant_name") or "").lower()
                      name = (raw.get("name") or "").lower()
                      counterparty_names = [
                          (cp.get("name") or "").lower()
                          for cp in (raw.get("counterparties") or [])
                      ]
                      if q in merchant or q in name or any(q in cp for cp in counterparty_names):
                          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:549-552 (registration)

  The search_transactions tool is registered with the MCP server using @mcp.tool with name='search_transactions', title='Search Transactions', and readOnlyHint=True. The implementation is _search_transactions_impl.

  search_transactions = mcp.tool(
      annotations={"readOnlyHint": True, "title": "Search Transactions"},
      name="search_transactions",
  )(_search_transactions_impl)

• server.py:119-129 (helper)

  The _clip_window helper clips the date range to a maximum lookback of ~2 years (730 days), returning clipped start/end dates and a warning reason if clipping occurred.

  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

• plaid_client.py:232-250 (helper)

  The shape_transaction helper transforms raw Plaid transaction dicts into trimmed response objects, dropping counterparty names and selecting personal_finance_category over legacy category.

  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")),
      }

• plaid_client.py:67-87 (helper)

  The map_plaid_error helper maps Plaid ApiExceptions to a standardized error dict with code, message, trace_id, and optional institution name.

  def map_plaid_error(exc: Exception, institution: str | None) -> dict:
      trace_id = str(uuid.uuid4())
      body: dict = {}
      try:
          parsed = json.loads(getattr(exc, "body", "") or "{}")
          body = parsed if isinstance(parsed, dict) else {}
      except (ValueError, TypeError):
          body = {}
      code = body.get("error_code") or body.get("error_type") or "UNKNOWN"
      message = body.get("error_message") or "Plaid call failed."
      request_id = body.get("request_id")
      _log.warning(
          "plaid_error trace_id=%s request_id=%s code=%s",
          trace_id,
          request_id,
          code,
      )
      err: dict = {"code": code, "message": message, "trace_id": trace_id}
      if institution:
          err["institution"] = institution
      return {"error": err}