You Need A Budget (YNAB) MCP

Instructions

Implementation Reference

• src/ynab_mcp_server/server.py:356-413 (handler)

  Main handler for the 'bulk-manage-transactions' tool. Validates input using BulkManageTransactionsInput, gets budget_id, then based on action ('create', 'update', 'delete'): prepares transaction data, calls corresponding ynab_client methods (create_transactions, update_transactions, delete_transaction), and returns formatted success response with details like created IDs or counts.
  elif name == "bulk-manage-transactions": args = BulkManageTransactionsInput.model_validate(arguments or {}) budget_id = await _get_budget_id(args.model_dump()) if args.action == "create": new_transactions = [] for tx in args.create_transactions: tx_data = {k: v for k, v in tx.model_dump().items() if v is not None} if "amount" in tx_data: tx_data["amount"] = int(tx_data["amount"]) new_transactions.append(NewTransaction(**tx_data)) result = await ynab_client.create_transactions( budget_id=budget_id, transactions=new_transactions ) created_ids = ", ".join(result.transaction_ids) duplicate_ids = ", ".join(result.duplicate_import_ids) response_text = f"Successfully processed bulk transaction request. Server Knowledge: {result.server_knowledge}\n" if created_ids: response_text += f"Created transaction IDs: {created_ids}\n" if duplicate_ids: response_text += f"Duplicate import IDs (skipped): {duplicate_ids}\n" return [types.TextContent(type="text", text=response_text.strip())] elif args.action == "update": updates = [] for tx in args.update_transactions: tx_data = {k: v for k, v in tx.model_dump().items() if v is not None} if "amount" in tx_data: tx_data["amount"] = int(tx_data["amount"]) updates.append(SaveTransactionWithIdOrImportId(**tx_data)) await ynab_client.update_transactions(budget_id=budget_id, transactions=updates) return [ types.TextContent( type="text", text=f"Successfully updated {len(args.update_transactions)} transactions.", ) ] elif args.action == "delete": deleted_ids = [] for tx_id in args.delete_transaction_ids: await ynab_client.delete_transaction( budget_id=budget_id, transaction_id=tx_id ) deleted_ids.append(tx_id) return [ types.TextContent( type="text", text=f"Successfully deleted {len(deleted_ids)} transactions: {', '.join(deleted_ids)}", ) ] elif name == "list-scheduled-transactions":
• src/ynab_mcp_server/tool_models.py:187-217 (schema)

  Pydantic input schema for bulk-manage-transactions tool, defining action enum and fields for create/update/delete operations. Includes model validator to enforce required fields based on action. References sub-models NewTransactionModel and TransactionUpdate.
  class BulkManageTransactionsAction(str, Enum): CREATE = "create" UPDATE = "update" DELETE = "delete" class BulkManageTransactionsInput(BudgetIdInput): action: BulkManageTransactionsAction = Field(..., description="The action to perform.") create_transactions: Optional[List[NewTransactionModel]] = Field(None, description="A list of transactions to create. Required for 'create' action.") update_transactions: Optional[List[TransactionUpdate]] = Field(None, description="A list of transactions to update. Required for 'update' action.") delete_transaction_ids: Optional[List[str]] = Field(None, description="A list of transaction IDs to delete. Required for 'delete' action.") @model_validator(mode='before') @classmethod def check_fields_for_action(cls, values): action = values.get('action') if not action: raise ValueError("'action' is a required field.") if action == 'create': if not values.get('create_transactions'): raise ValueError("'create_transactions' is required for the 'create' action.") elif action == 'update': if not values.get('update_transactions'): raise ValueError("'update_transactions' is required for the 'update' action.") elif action == 'delete': if not values.get('delete_transaction_ids'): raise ValueError("'delete_transaction_ids' is required for the 'delete' action.") return values
• src/ynab_mcp_server/server.py:87-91 (registration)

  Tool registration in the handle_list_tools() function, defining the tool's name, description, and input schema from BulkManageTransactionsInput.model_json_schema().
  types.Tool( name="bulk-manage-transactions", description="Create, update, or delete multiple transactions at once. More efficient than making single changes.", inputSchema=BulkManageTransactionsInput.model_json_schema(), ),
• src/ynab_mcp_server/tool_models.py:144-166 (schema)

  Sub-schema NewTransactionModel used in create_transactions list of BulkManageTransactionsInput for defining new transaction details.
  class NewTransactionModel(BaseModel): account_id: str = Field(..., description="The ID of the account for the transaction.") date: str = Field(..., description="The transaction date in YYYY-MM-DD format.") amount: float = Field(..., description="The transaction amount in milliunits.") payee_id: Optional[str] = Field(None, description="The ID of the payee.") payee_name: Optional[str] = Field( None, description="The name of the payee. If not provided, a new payee will be created." ) category_id: Optional[str] = Field( None, description="The ID of the category for the transaction." ) memo: Optional[str] = Field(None, description="A memo for the transaction.") cleared: Optional[str] = Field( None, description="The cleared status of the transaction.", ) approved: bool = Field(False, description="Whether or not the transaction is approved.") flag_color: Optional[str] = Field( None, description="The flag color of the transaction.", ) import_id: Optional[str] = Field( None, description="A unique import ID for the transaction. Use for idempotency." )
• src/ynab_mcp_server/tool_models.py:168-185 (schema)

  Sub-schema TransactionUpdate used in update_transactions list of BulkManageTransactionsInput for defining updates to existing transactions.
  class TransactionUpdate(BaseModel): id: str = Field(..., description="The ID of the transaction to update.") account_id: Optional[str] = Field(None, description="The ID of the account.") date: Optional[str] = Field(None, description="The transaction date in YYYY-MM-DD format.") amount: Optional[float] = Field(None, description="The transaction amount in milliunits.") category_id: Optional[str] = Field( None, description="The ID of the category for the transaction." ) payee_id: Optional[str] = Field(None, description="The ID of the payee.") memo: Optional[str] = Field(None, description="A memo for the transaction.") cleared: Optional[str] = Field( None, description="The cleared status of the transaction.", ) approved: Optional[bool] = Field(None, description="Whether or not the transaction is approved.") flag_color: Optional[str] = Field( None, description="The flag color of the transaction.", )