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