get_scheduled_transactions
Retrieve all scheduled transactions from your YNAB budget to manage upcoming payments and maintain financial planning accuracy.
Instructions
Get all scheduled transactions.
Args:
budget_id: The ID of the budget (use 'last-used' for default budget)
Returns:
JSON string with list of scheduled transactionsInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| budget_id | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- src/ynab_mcp/server.py:343-356 (handler)MCP tool handler for get_scheduled_transactions. Registers the tool and delegates to YNABClient instance.
@mcp.tool() async def get_scheduled_transactions(budget_id: str) -> str: """Get all scheduled transactions. Args: budget_id: The ID of the budget (use 'last-used' for default budget) Returns: JSON string with list of scheduled transactions """ client = get_ynab_client() result = await client.get_scheduled_transactions(budget_id) return json.dumps(result, indent=2) - Core implementation in YNABClient that fetches scheduled transactions from YNAB API, formats the response, and handles errors.
async def get_scheduled_transactions(self, budget_id: str) -> list[dict[str, Any]]: """Get all scheduled transactions. Args: budget_id: The budget ID or 'last-used' Returns: List of scheduled transaction dictionaries """ try: url = f"{self.api_base_url}/budgets/{budget_id}/scheduled_transactions" result = await self._make_request_with_retry("get", url) scheduled_txns = [] for txn in result["data"]["scheduled_transactions"]: scheduled_txns.append( { "id": txn["id"], "date_first": txn.get("date_first"), "date_next": txn.get("date_next"), "frequency": txn.get("frequency"), "amount": txn["amount"] / 1000 if txn.get("amount") else 0, "memo": txn.get("memo"), "flag_color": txn.get("flag_color"), "account_id": txn.get("account_id"), "account_name": txn.get("account_name"), "payee_id": txn.get("payee_id"), "payee_name": txn.get("payee_name"), "category_id": txn.get("category_id"), "category_name": txn.get("category_name"), "deleted": txn.get("deleted"), } ) return scheduled_txns except Exception as e: raise Exception(f"Failed to get scheduled transactions: {e}") from e